diff --git a/Dockerfile.amqp-adapter b/Dockerfile.amqp-adapter index 4db82e3..cc0ec82 100644 --- a/Dockerfile.amqp-adapter +++ b/Dockerfile.amqp-adapter @@ -11,17 +11,10 @@ COPY ./requirements.txt . # Install the Python dependencies RUN pip install --no-cache-dir -r requirements.txt -run pip install "fastapi[standard]" # Copy the application code into the container COPY ./.env_secrets . COPY ./amqp ./amqp COPY ./container-data ./container-data -# Expose the port that the FastAPI app will run on -EXPOSE 8080 - -# Set the command to run the FastAPI app CMD ["python", "amqp/service/amq_service.py"] -#CMD ["fastapi", "run", "amqp/service/amq_service.py", "--host", "0.0.0.0", "--port", "8080"] -#CMD ["sleep", "300"] diff --git a/amqp/adapter/amq_message_factory.py b/amqp/adapter/amq_message_factory.py index ae08561..2a285d6 100644 --- a/amqp/adapter/amq_message_factory.py +++ b/amqp/adapter/amq_message_factory.py @@ -13,7 +13,7 @@ from amqp.router.utils import sanitize_as_rabbitmq_name class AMQMessageFactory: SNOWFLAKE_ID_BITS = 80 SEQUENCE_BITS = 12 - MACHINE_ID_BITS = 26 + MACHINE_ID_BITS = 24 TIMESTAMP_BITS = SNOWFLAKE_ID_BITS - SEQUENCE_BITS - MACHINE_ID_BITS MACHINE_ID_MASK = (1 << MACHINE_ID_BITS) - 1 @@ -30,11 +30,20 @@ class AMQMessageFactory: @classmethod def get_instance(cls, machine_id: int): + """ + return reusable factory instance + :param machine_id: a generator ID, should be unique for each component to prevent duplicate IDs + :return: factory instance + """ if cls._instance is None: cls._instance = AMQMessageFactory(machine_id) return cls._instance def generate_snowflake_id(self) -> SnowflakeId: + """ + Generate an unique ID following the SnowflakeID structure, see https://en.wikipedia.org/wiki/Snowflake_ID + :return: 80-bit wide ID + """ timestamp = int(datetime.now(timezone.utc).timestamp() * 1000) if timestamp < self.snowflake_last_timestamp: @@ -64,6 +73,15 @@ class AMQMessageFactory: headers: Dict[str, List[str]], message, ) -> AMQMessage: + """ + Encapsulate / hide the system level inputs, create the AMQ message from business relevant input + :param method: a method requested (e.g. HTTP method like GET, POST in case of REST request) + :param domain: service host name + :param path: path at the service + :param headers: HTTP headers + :param message: payload + :return: AMQMessage object + """ serializable_headers = headers.copy() if headers else {} payload = base64.b64encode(message.encode('utf-8') if isinstance(message, str) else message) trace_info = {} @@ -80,10 +98,20 @@ class AMQMessageFactory: @staticmethod def amq_message_routing_key(message: AMQMessage) -> str: + """ + Constructs the message specific routing key. + :param message: AMQMessage input + :return: routing key (as string) compliant to RabbitMQ allowed character set. + """ return sanitize_as_rabbitmq_name(f"{message.domain}.{message.path}") @staticmethod def get_timestamp_from_id(_id: SnowflakeId) -> int: + """ + Helper method - retrieve the timestamp from the ID + :param _id: SnowflakeID (includes encoded timestamp) + :return: the message timestamp + """ bits = _id.get_bits() low = bits[0] >> (AMQMessageFactory.SEQUENCE_BITS + AMQMessageFactory.MACHINE_ID_BITS) high = bits[1] << (64 - AMQMessageFactory.SEQUENCE_BITS - AMQMessageFactory.MACHINE_ID_BITS) @@ -91,6 +119,11 @@ class AMQMessageFactory: @staticmethod def to_string(message: AMQMessage) -> str: + """ + Human-readable representation. + :param message: AMQMessage input + :return: as string + """ return ( f"AMQMessage{{" f"id={message.id}, " @@ -105,6 +138,11 @@ class AMQMessageFactory: @staticmethod def serialize(message: AMQMessage) -> bytes: + """ + Converts AMQMessage to byte array + :param message: message to serialize + :return: messages as bytes + """ id_bytes = str(message.id).encode() prolog = ( f"|m={message.method}|d={message.domain}|p={message.path}|h={{{map_with_list_as_string(message.headers)}}}|t={{{map_as_string(message.trace_info)}}}|b=" @@ -122,6 +160,11 @@ class AMQMessageFactory: @staticmethod def from_bytes(input_bytes: bytes) -> AMQMessage: + """ + Builds the AMQMessage from its serialized (byte array) form. + :param input_bytes: serialized message + :return: AMQMessage instance + """ prolog_len = (input_bytes[0] << 8) + input_bytes[1] metadata = input_bytes[2:prolog_len + 2].decode() diff --git a/amqp/adapter/amq_response_factory.py b/amqp/adapter/amq_response_factory.py index d8dd14c..59391cb 100644 --- a/amqp/adapter/amq_response_factory.py +++ b/amqp/adapter/amq_response_factory.py @@ -11,12 +11,26 @@ class AMQResponseFactory: @staticmethod def create_async_response_message(_id: SnowflakeId) -> AMQResponse: + """ + Helper method to create standard OK response. + :param _id: ID + :return: Response message + """ return AMQResponse( str(_id), 200, "application/text", "OK".encode("utf-8"), "", "", {} ) @staticmethod def of(id: SnowflakeId, response_code: int, content_type: str, body: bytes, trace_info: dict[str,str]) -> AMQResponse: + """ + Create the AMQResponse message supplying all key values + :param id: SnowflakeID + :param response_code: HTTP response code + :param content_type: MIME content type + :param body: payload + :param trace_info: Jaeger trace info (id) + :return: AMQResponseMessage object + """ return AMQResponse( id=id.as_string(), response_code=response_code, @@ -29,6 +43,11 @@ class AMQResponseFactory: @staticmethod def serialize(response: AMQResponse) -> bytes: + """ + Serialize the message to byte array (to be sent over RabbitMQ) + :param response: Object to serialize + :return: byte array as serialized message + """ id_bytes = response.id.encode('utf-8') prolog = ( f"|r={response.response_code}|c={response.content_type}|e={response.error}|f={response.error_cause}|t={{{map_as_string(response.trace_info)}}}|b=" @@ -46,6 +65,11 @@ class AMQResponseFactory: @staticmethod def from_bytes(input_bytes: bytes) -> AMQResponse: + """ + Buil;ds the AMQResponse object from its serialized form + :param input_bytes: byte array - serialized response + :return: AMQResponse object + """ prolog_len = (input_bytes[0] << 8) + input_bytes[1] metadata = input_bytes[2:prolog_len + 2].decode() diff --git a/amqp/adapter/amq_route_factory.py b/amqp/adapter/amq_route_factory.py index acf48e4..aa1afe2 100644 --- a/amqp/adapter/amq_route_factory.py +++ b/amqp/adapter/amq_route_factory.py @@ -17,6 +17,11 @@ class AMQRouteFactory: @staticmethod def from_string(data): + """ + Builds the AMQRoute from its string form. + :param data: route string + :return: AMQRoute object + """ try: keys = data.split(RS) if len(keys) >= 5: @@ -55,6 +60,11 @@ class AMQRouteFactory: @staticmethod def validate(route_str): + """ + Validates the route string. + :param route_str: route string + :return: boolean validity indicator + """ if route_str is not None: return route_str.count(":") > 4 return False diff --git a/amqp/adapter/clever_swarm_adapter.py b/amqp/adapter/clever_swarm_adapter.py index 7fd8667..bda8fb2 100644 --- a/amqp/adapter/clever_swarm_adapter.py +++ b/amqp/adapter/clever_swarm_adapter.py @@ -1,10 +1,200 @@ import logging -from typing import Dict, List -from aiohttp import ClientSession, ClientResponse +import re +from typing import Dict, List, Tuple, Callable, Coroutine, AnyStr, Any +import python_multipart +from aiohttp import ClientSession, ClientResponse +from fastapi import HTTPException, UploadFile +from python_multipart.multipart import Field, File + +from amqp.adapter.amq_response_factory import AMQResponseFactory from amqp.config.amq_configuration import AMQConfiguration from amqp.model.model import AMQMessage, AMQResponse +import rest.v0.endpoints.benchmark as benchmark +import rest.v0.endpoints.jobs as jobs +import rest.v0.endpoints.unstructured as unstructured +from core.deps import get_current_user +from schemas.user_schema import UserSchema + +import json +from urllib.parse import parse_qs +from xml.etree import ElementTree +import io + + +class CleverMultiPartParser: + def __init__(self, message: AMQMessage): + self.form_data = {} + # Convert each list of strings to a single string joined by newline, then to bytes + headers = {key: "\n".join(value).encode("utf-8") for key, value in message.headers.items()} + python_multipart.parse_form(headers, io.BytesIO(message.body()), self.on_field, self.on_file) + + def on_field(self, field: Field): + print(field) + self.form_data[field.field_name.decode('utf-8')] = field.value + print(self.form_data) + + def on_file(self, file: File): + print(file) + self.form_data[file.field_name.decode('utf-8')] = \ + UploadFile(file.file_object, size=file.size, filename=file.file_name) + print(self.form_data) + + +def parse_request_body(message: AMQMessage): + """ + Parses the HTTP POST request body based on the MIME type. + + Args: + body (str): The raw request body as a string. + content_type (str): The Content-Type header of the request. + + Returns: + dict: Parsed data as a dictionary. + """ + content_type = message.headers.get('Content-Type', '') + if not content_type: + raise ValueError("Content-Type header is required") + + # Parse JSON + if content_type == "application/json": + try: + return json.loads(message.body()) + except json.JSONDecodeError as e: + raise ValueError(f"Invalid JSON: {e}") + + # Parse URL-encoded form data + elif content_type == "application/x-www-form-urlencoded": + try: + return {k: v[0] if len(v) == 1 else v for k, v in parse_qs(message.body()).items()} + except Exception as e: + raise ValueError(f"Invalid URL-encoded form data: {e}") + + # Parse multipart/form-data + elif content_type.startswith("multipart/form-data"): + try: + parser: CleverMultiPartParser = CleverMultiPartParser(message) + return parser.form_data + except Exception as e: + raise ValueError(f"Invalid multipart/form-data: {e}") + + # Parse plain text + elif content_type == "text/plain": + return {"text": message.body()} + + # Parse XML + elif content_type == "application/xml": + try: + root = ElementTree.fromstring(message.body()) + return {elem.tag: elem.text for elem in root} + except ElementTree.ParseError as e: + raise ValueError(f"Invalid XML: {e}") + + else: + raise ValueError(f"Unsupported Content-Type: {content_type}") + + +BENCHMARK = '/benchmark' +JOBS = '/jobs/' +UNSTRUCTURED = '/unstructured' + +GET_BENCHMARK_KG = r".*/benchmark/(?P[^\?]+)\?index=(?P\d+)" +GET_JOB_STATUS = r".*/jobs/(?P.*)" +GET_KG = r".*/unstructured/(?P.*)" +DUMMY = r"(?P.*)" + + +# ============= POST Helper function ============= + +async def call_cleverswarm_post_put_api( + message: AMQMessage, + args: Any, + api_method: Callable[[Any, UserSchema], Coroutine], + success_code: int, + authenticated_user: UserSchema +) -> AMQResponse: + try: + coroutine_call: Coroutine = api_method(args, authenticated_user) + result: str = await coroutine_call + return AMQResponseFactory.of( + message.id, success_code, message.headers.get("Content-Type", "application/json"), + result.encode('utf-8'), message.trace_info + ) + except HTTPException as http_error: + return AMQResponseFactory.of(message.id, http_error.status_code, "text/plain", + str(http_error.detail).encode('utf-8'), message.trace_info) + except Exception as error: + return AMQResponseFactory.of(message.id, 500, "text/plain", str(error).encode('utf-8'), message.trace_info) + + +# +# Following methods are wrappers to actual CleverSwarm endpoints, so I can refer to the CleverSwarm method and +# pass it as input parameter using general Callable[[Any, UserSchema], str] type +# +async def cleverswarm_create_benchmark_job(args: Any, authenticated_user: UserSchema) -> str: + return str(await benchmark.create_benchmark_job(**args, authenticated_user=authenticated_user)) + + +async def cleverswarm_create_extract_with_ontology(args: Any, authenticated_user: UserSchema) -> str: + return str(await unstructured.create_extract_with_ontology(**args, authenticated_user=authenticated_user)) + + +async def cleverswarm_create_extract_with_wildcards(args: Any, authenticated_user: UserSchema) -> str: + return str(await unstructured.create_extract_with_ontology(**args, authenticated_user=authenticated_user)) + + +async def cleverswarm_update_benchmark_job(args: Any, authenticated_user: UserSchema) -> str: + return str(await benchmark.update_benchmark_job(**args, authenticated_user=authenticated_user)) + + +async def cleverswarm_retry_job(args: Any, authenticated_user: UserSchema) -> str: + return str(await jobs.retry_job(args[0], authenticated_user=authenticated_user)) + + +async def cleverswarm_delete_job(args: Any, authenticated_user: UserSchema) -> str: + return str(await jobs.delete_job(args[0], authenticated_user=authenticated_user)) + + +# ============= GET Helper function ============= +async def call_cleverswarm_get_api( + message: AMQMessage, + pattern: str, + api_method: Callable[[Tuple[AnyStr | Any, ...], UserSchema], Coroutine], + authenticated_user: UserSchema +) -> AMQResponse: + try: + match = re.search(pattern, message.path) + if match: + crt: Coroutine = api_method(match.groups(), authenticated_user) + result: str = await crt + return AMQResponseFactory.of(message.id, 200, "text/plain", result.encode('utf-8'), message.trace_info) + else: + _msg = f"Unexpected URL: {message.path}, expected pattern: {pattern}" + return AMQResponseFactory.of(message.id, 404, "text/plain", _msg.encode('utf-8'), message.trace_info) + except HTTPException as http_error: + return AMQResponseFactory.of(message.id, http_error.status_code, "text/plain", + str(http_error.detail).encode('utf-8'), message.trace_info) + except Exception as error: + return AMQResponseFactory.of(message.id, 500, "text/plain", str(error).encode('utf-8'), message.trace_info) + + +async def cleverswarm_get_benchmark_kg(args: Tuple[AnyStr | Any, ...], authenticated_user: UserSchema) -> str: + return str(await benchmark.get_benchmark_kg(args[0], args[1], authenticated_user=authenticated_user)) + + +async def cleverswarm_get_job_status(args: Tuple[AnyStr | Any, ...], authenticated_user: UserSchema) -> str: + return str(await jobs.get_job_status(args[0], authenticated_user=authenticated_user)) + + +async def cleverswarm_jobs_get_requests(args: Tuple[AnyStr | Any, ...], + authenticated_user: UserSchema) -> str: + return str(await jobs.get_requests(authenticated_user=authenticated_user)) + + +async def cleverswarm_get_kg(args: Tuple[AnyStr | Any, ...], authenticated_user: UserSchema) -> str: + return str(await unstructured.get_kg(args[0], authenticated_user=authenticated_user)) + class CleverSwarmAdapter: def __init__(self, amq_configuration: AMQConfiguration, session: ClientSession = None): @@ -13,55 +203,181 @@ class CleverSwarmAdapter: self.service_host = self.amq_configuration.dispatch.service_host self.service_port = self.amq_configuration.dispatch.service_port - def on_message(self, amq_message) -> AMQResponse: - pass - - async def handle(self, message: AMQMessage) -> ClientResponse: - method = message.method.lower() - if method == 'get': - return await self.get(message) - elif method == 'put': - return await self.put(message) - elif method == 'post': - return await self.post(message) - elif method == 'head': - return await self.head(message) - elif method == 'delete': - return await self.delete(message) + async def on_message(self, message) -> AMQResponse: + DEBUG_NO_AUTH = 1 + _auth: str = message.headers.get('Authorization', '') + _auth_user: UserSchema | None = None + if 'Bearer' in _auth: + try: + token = _auth.split(' ')[1] + _auth_user = await get_current_user(token) + except Exception as error: + logging.error(f"Error getting user: {error}") + if _auth_user or DEBUG_NO_AUTH == 1: + method = message.method.lower() + if method == 'get': + return await self.get(message, _auth_user) + elif method == 'put': + return await self.put(message, _auth_user) + elif method == 'post': + return await self.post(message, _auth_user) + elif method == 'head': + return await self.head(message, _auth_user) + elif method == 'delete': + return await self.delete(message, _auth_user) + else: + raise ValueError(f"Unexpected HTTP method: {message.method}") else: - raise ValueError(f"Unexpected HTTP method: {message.method}") + return AMQResponseFactory.of(message.id, 401, "text/plain", b"Unauthorized", message.trace_info) - async def get(self, message: AMQMessage) -> ClientResponse: + # ================================================================================================== + # ===================== C l e v e r S w a r m A P I m e t h o d s ============================ + # ================================================================================================== + + async def get(self, message: AMQMessage, auth_user: UserSchema) -> AMQResponse: + """ + Handles the GET requests for the CleverSwarm API. + :param message: message from the AMQP that contains the GET request + :param auth_user: user authenticated by the API Gateway, as UserSchem + :return: AMQResponse to be sent back to the API Gateway via AMQP + """ + if BENCHMARK in message.path: + return await call_cleverswarm_get_api( + message, + GET_BENCHMARK_KG, + cleverswarm_get_benchmark_kg, + authenticated_user=auth_user + ) + + if JOBS in message.path: + return await call_cleverswarm_get_api( + message, + GET_JOB_STATUS, + cleverswarm_get_job_status, + authenticated_user=auth_user + ) + + if message.path.endswith('/jobs'): + return await call_cleverswarm_get_api( + message, + DUMMY, + cleverswarm_jobs_get_requests, + authenticated_user=auth_user + ) + + if UNSTRUCTURED in message.path: + return await call_cleverswarm_get_api( + message, + GET_KG, + cleverswarm_get_kg, + authenticated_user=auth_user + ) + + # unmatched path - try to invoke the service directly on this path url = f"http://{self.service_host}:{self.service_port}{message.path}" headers = {**message.headers, **message.trace_info} logging.info(f" [*] SEND HEADERS out: {', '.join(f'{k}={v}' for k, v in headers.items())}, REQ={url}") - return await self.session.get(url, headers=headers) + if self.session: + _http_resp: ClientResponse = await self.session.get(url, headers=headers) + return AMQResponseFactory.of(message.id, _http_resp.status, _http_resp.content_type, + await _http_resp.read(), + message.trace_info) + return AMQResponseFactory.of(message.id, 404, "text/plain", + str("Destination location does not exists").encode('utf-8'), message.trace_info) + + async def put(self, message: AMQMessage, auth_user: UserSchema) -> AMQResponse: + """ + Handles the PUT requests for the CleverSwarm API. + :param message: message from the AMQP that contains the PUT request + :param auth_user: user authenticated by the API Gateway, as UserSchem + :return: AMQResponse to be sent back to the API Gateway via AMQP + """ + parser: CleverMultiPartParser = CleverMultiPartParser(message) + if BENCHMARK in message.path: + return await call_cleverswarm_post_put_api( + message, + parser.form_data, + cleverswarm_create_benchmark_job, + 201, + authenticated_user=auth_user + ) + + if JOBS in message.path: + return await call_cleverswarm_get_api( + message, + GET_JOB_STATUS, + cleverswarm_retry_job, + authenticated_user=auth_user + ) - async def put(self, message: AMQMessage) -> ClientResponse: return await self.handle_possible_form( self.session.put( f"http://{self.service_host}:{self.service_port}{message.path}", data=message.body, headers={**message.headers, **message.trace_info}, - ) + ), + auth_user ) - async def post(self, message: AMQMessage) -> ClientResponse: + async def post(self, message: AMQMessage, auth_user: UserSchema) -> AMQResponse: + """ + Handles the POST requests for the CleverSwarm API. + :param message: message from the AMQP that contains the POST request, including body and headers + :param auth_user: user authenticated by the API Gateway, as UserSchem + :return: AMQResponse to be sent back to the API Gateway via AMQP + """ + parser: CleverMultiPartParser = CleverMultiPartParser(message) + if BENCHMARK in message.path: + return await call_cleverswarm_post_put_api( + message, + parser.form_data, + cleverswarm_create_benchmark_job, + 201, + authenticated_user=auth_user + ) + if UNSTRUCTURED in message.path: + if 'with_ontology' in message.path: + return await call_cleverswarm_post_put_api( + message, + parser.form_data, + cleverswarm_create_extract_with_ontology, + 200, + authenticated_user=auth_user + ) + if 'with_wildcards' in message.path: + return await call_cleverswarm_post_put_api( + message, + parser.form_data, + cleverswarm_create_extract_with_wildcards, + 200, + authenticated_user=auth_user + ) + return await self.handle_possible_form( self.session.post( f"http://{self.service_host}:{self.service_port}{message.path}", data=message.body, headers={**message.headers, **message.trace_info}, - ) + ), + auth_user ) - async def handle_possible_form(self, request_coroutine) -> ClientResponse: + async def handle_possible_form(self, request_coroutine, auth_user: UserSchema) -> AMQResponse: return await request_coroutine - async def head(self, message: AMQMessage) -> ClientResponse: + async def head(self, message: AMQMessage, auth_user: UserSchema) -> AMQResponse: raise NotImplementedError - async def delete(self, message: AMQMessage) -> ClientResponse: + async def delete(self, message: AMQMessage, auth_user: UserSchema) -> AMQResponse: + parser: CleverMultiPartParser = CleverMultiPartParser(message) + if JOBS in message.path: + return await call_cleverswarm_get_api( + message, + GET_JOB_STATUS, + cleverswarm_retry_job, + authenticated_user=auth_user + ) + raise NotImplementedError def get_content_type(self, message_headers: Dict[str, List[str]]) -> str: diff --git a/amqp/adapter/response_holder.py b/amqp/adapter/response_holder.py deleted file mode 100644 index 1e5538b..0000000 --- a/amqp/adapter/response_holder.py +++ /dev/null @@ -1,15 +0,0 @@ -from dataclasses import dataclass -from concurrent.futures import Future - -from aiohttp.web_response import Response -from opentelemetry.trace import Span - - -@dataclass -class ResponseHolder: - response: Future[Response] - tracing_span: Span - - def __init__(self, response: Future[Response], span: Span): - self.response = response - self.tracing_span = span diff --git a/amqp/adapter/service_adapter.py b/amqp/adapter/service_adapter.py deleted file mode 100644 index 1454951..0000000 --- a/amqp/adapter/service_adapter.py +++ /dev/null @@ -1,5 +0,0 @@ -class ServiceAdapter: - - def __init__(self): - pass - diff --git a/amqp/adapter/service_message_factory.py b/amqp/adapter/service_message_factory.py index cabf678..2e8e3d8 100644 --- a/amqp/adapter/service_message_factory.py +++ b/amqp/adapter/service_message_factory.py @@ -16,11 +16,18 @@ INVALID_MESSAGE = CleverMicroServiceMessage() # default values mean invalid mes class CleverMicroServiceMessageFactory: - + """ + build/serialize/deserialize CleverMicro Service Message class + """ def __init__(self, name: str): self.process_name = name def to_string(self, message: CleverMicroServiceMessage) -> str: + """ + Convert CleverMicroServiceMessage to human-readable format + :param message: CleverMicroServiceMessage object + :return: as string + """ if message is not None: return ( f"CleverMicroServiceMessage{{id={message.id.as_string()}|type={self.format_message_type(message)}|" @@ -30,6 +37,13 @@ class CleverMicroServiceMessageFactory: return "null" def of(self, message_type: ServiceMessageType, payload: str, recipient_name: str) -> CleverMicroServiceMessage: + """ + Builds CleverMicroServiceMessage object from relevant values. + :param message_type: enum Message type + :param payload: Message content, stored as-is + :param recipient_name: Recipient name (from the routing expression) + :return: CleverMicroServiceMessage object + """ _id = AMQMessageFactory.get_instance(1).generate_snowflake_id() return CleverMicroServiceMessage( id=_id, @@ -42,6 +56,13 @@ class CleverMicroServiceMessageFactory: ) def as_base64(self, message_type: ServiceMessageType, payload: str, recipient_name: str) -> CleverMicroServiceMessage: + """ + Builds CleverMicroServiceMessage object from relevant values, stores content Base64 encoded. + :param message_type: enum Message type + :param payload: Message content, stored as Base64 string + :param recipient_name: Recipient name (from the routing expression) + :return: CleverMicroServiceMessage object + """ _id = AMQMessageFactory.get_instance(1).generate_snowflake_id() return CleverMicroServiceMessage( id=_id, @@ -54,6 +75,11 @@ class CleverMicroServiceMessageFactory: ) def from_bytes(self, input_bytes: bytes) -> CleverMicroServiceMessage: + """ + Deserialize CleverMicroServiceMessage from byte array serialized form. + :param input_bytes: serialized CleverMicroServiceMessage + :return: CleverMicroServiceMessage object represented by the input. + """ logging.info(f"CleverMicroServiceMessage >>> [{input_bytes.decode()}]") input_str_array = input_bytes.decode().split(RS) try: @@ -89,6 +115,11 @@ class CleverMicroServiceMessageFactory: return INVALID_MESSAGE def serialize(self, message: CleverMicroServiceMessage) -> bytes: + """ + Serialize CleverMicroServiceMessage to byte array. + :param message: CleverMicroServiceMessage to serialize. + :return: serialized object. + """ return ( f"{message.id.as_string()}{RS}{self.format_message_type(message)}{RS}{message.recipient_name}{RS}" f"{message.reply_to}{RS}{map_as_string(message.trace_info)}{RS}" @@ -98,7 +129,7 @@ class CleverMicroServiceMessageFactory: def format_message_type(self, message: CleverMicroServiceMessage) -> str: """ - Formats the Enum message type and ecorates it with base64 flag at highest bit. + Formats the Enum message type and decorates it with base64 flag at highest bit. For compatibility with 0-based enum ordinals in Java subtract 1 from ordinal value. :param message: Service message to send :return: formatted message type diff --git a/amqp/adapter/test_amq_message_factory.py b/amqp/adapter/test_amq_message_factory.py index 8a782c0..f424d5a 100644 --- a/amqp/adapter/test_amq_message_factory.py +++ b/amqp/adapter/test_amq_message_factory.py @@ -53,7 +53,7 @@ class AMQMessageFactoryTest(TestCase): _f = AMQMessageFactory(35) _req = _f.create_request_message("PUT", "localhost", "/test/me", {"k1": ["v1"], "k2": ["v2"]}, "Quick brown fox") _key = _f.amq_message_routing_key(_req) - self.assertEqual("localhost..test.me", _key, "Message routing key wrongly extracted from the message") + self.assertEqual("localhost.test.me", _key, "Message routing key wrongly extracted from the message") def test_get_timestamp_from_id(self): _f = AMQMessageFactory(35) diff --git a/amqp/adapter/test_service_message_factory.py b/amqp/adapter/test_service_message_factory.py index 44ab7fe..c049622 100644 --- a/amqp/adapter/test_service_message_factory.py +++ b/amqp/adapter/test_service_message_factory.py @@ -1,6 +1,6 @@ from unittest import TestCase -from amqp.adapter.service_message_factory import CleverMicroServiceMessageFactory, RS +from amqp.adapter.service_message_factory import CleverMicroServiceMessageFactory, RS, INVALID_MESSAGE from amqp.model.model import CleverMicroServiceMessage from amqp.model.service_message_type import ServiceMessageType @@ -11,13 +11,15 @@ class CleverMicroServiceMessageTest(TestCase): _f: CleverMicroServiceMessageFactory = CleverMicroServiceMessageFactory('bumble-bee') message: CleverMicroServiceMessage = _f.of(ServiceMessageType.ROUTING_DATA, "Duck", "Donald") message_str = _f.to_string(message) - assert "|type=01|" in message_str - assert "|body=Duck" in message_str + self.assertTrue("|type=01|" in message_str) + self.assertTrue("|body=Duck" in message_str) message: CleverMicroServiceMessage = _f.of(ServiceMessageType.ROUTING_DATA_REPLACE, "Duck", "Donald") message_str = _f.to_string(message) - assert "|type=03|" in message_str - assert "|body=Duck" in message_str - assert "|replyTo=bumble-bee" in message_str + self.assertTrue("|type=03|" in message_str) + self.assertTrue("|body=Duck" in message_str) + self.assertTrue("|replyTo=bumble-bee" in message_str) + nullMsg = _f.to_string(None) + self.assertEqual("null", nullMsg) def test_from(self): _f: CleverMicroServiceMessageFactory = CleverMicroServiceMessageFactory("amqp-router") @@ -29,6 +31,8 @@ class CleverMicroServiceMessageTest(TestCase): assert deserialized.recipient_name == "Donald" assert deserialized.message == "Duck" assert deserialized.reply_to == "amqp-router" + invalidMsg = _f.from_bytes(b'wrong') + self.assertEqual(invalidMsg, INVALID_MESSAGE) def test_base64(self): _f: CleverMicroServiceMessageFactory = CleverMicroServiceMessageFactory("amqp-router") diff --git a/amqp/adapter/trace_info_adapter.py b/amqp/adapter/trace_info_adapter.py index d4650f7..b152bd8 100644 --- a/amqp/adapter/trace_info_adapter.py +++ b/amqp/adapter/trace_info_adapter.py @@ -4,7 +4,7 @@ from typing import Dict @dataclass class TraceInfoAdapter: """ - This class should set up the Open Telemetry tracing stack. To be done later. + TODO This class should set up the Open Telemetry tracing stack. To be done later. """ @staticmethod diff --git a/amqp/config/amq_configuration.py b/amqp/config/amq_configuration.py index 343183e..f98b4e2 100644 --- a/amqp/config/amq_configuration.py +++ b/amqp/config/amq_configuration.py @@ -1,8 +1,15 @@ import configparser import os +""" +Convert configuration from properties file to an object. +""" + class AMQAdapter: + """ + configuration with prefix 'cm.amq-adapter' + """ PREFIX: str = 'cm.amq-adapter.' @@ -23,8 +30,10 @@ class AMQAdapter: return f"{self.service_name}-{self.generator_id}-" - class Dispatch: + """ + configuration with prefix 'cm.dispatch' + """ PREFIX: str = 'cm.dispatch.' @@ -52,6 +61,9 @@ class Dispatch: class Backpressure: + """ + configuration with prefix 'cm.backpressure' + """ PREFIX: str = 'cm.backpressure.' @@ -68,15 +80,28 @@ class Backpressure: self.time_window = config.getint('CleverMicro-AMQ', self.PREFIX + 'time-window', fallback=3000) -class AMQConfiguration: +DEFAULT_PROPERTIES_FILE = '/app/amqp/config/application.properties' + +class AMQConfiguration: + """ + Top level configuration context holder / object + """ def __init__(self): config = configparser.ConfigParser() # Read the application.properties file - if os.getenv('USER', 'linux') == 'stanhejny': - config.read('../config/application.properties.local') + if os.path.exists(DEFAULT_PROPERTIES_FILE): + config.read(DEFAULT_PROPERTIES_FILE) else: - config.read('/app/amqp/config/application.properties') + config.read('../config/application.properties.local') + + # Override values with environment variables + for section in config.sections(): + for option in config.options(section): + # env_var = f"{section.upper()}_{option.upper()}" + if option in os.environ: + config.set(section, option, os.environ[option]) + # # Add individual config areas # diff --git a/amqp/config/application.properties b/amqp/config/application.properties index 40f7f28..607b570 100644 --- a/amqp/config/application.properties +++ b/amqp/config/application.properties @@ -5,7 +5,7 @@ cm.amq-adapter.service-name=amq-adapter cm.amq-adapter.is-router=false cm.amq-adapter.generator-id=164 -cm.amq-adapter.route-mapping=amq-clever-pybook::amq-clever-pybook:clever.pybook.localhost.#:5:0 +cm.amq-adapter.route-mapping=amq-cleverswarm::amq-cleverswarm:cleverswarm.localhost.#:5:0 cm.dispatch.use-dlq=true cm.dispatch.use-confirms=false diff --git a/amqp/config/application.properties.local b/amqp/config/application.properties.local index 10ded7d..6b8a65a 100644 --- a/amqp/config/application.properties.local +++ b/amqp/config/application.properties.local @@ -5,7 +5,7 @@ cm.amq-adapter.service-name=amq-adapter cm.amq-adapter.is-router=false cm.amq-adapter.generator-id=211 -cm.amq-adapter.route-mapping=amq-clever-pybook::amq-clever-pybook:clever.pybook.localhost.#:5:0 +cm.amq-adapter.route-mapping=amq-cleverswarm::amq-cleverswarm:cleverswarm.localhost.#:5:0 cm.dispatch.use-dlq=true cm.dispatch.use-confirms=false diff --git a/amqp/model/model.py b/amqp/model/model.py index d56a53e..d192615 100644 --- a/amqp/model/model.py +++ b/amqp/model/model.py @@ -10,6 +10,11 @@ from amqp.router.utils import sanitize_as_rabbitmq_name RS = ":" +""" +Define the structure of messages used in Clever Microś +""" + + @dataclass(frozen=True) class AMQRoute: """ @@ -47,6 +52,10 @@ class AMQRoute: @property def primary_key(self): + """ + Build the unique identifier of the route + :return primary key + """ return f"{self.component_name}{RS}{self.queue}{RS}{self.key}" def as_string(self) -> str: diff --git a/amqp/model/snowflake_id.py b/amqp/model/snowflake_id.py index cdfab3b..655da64 100644 --- a/amqp/model/snowflake_id.py +++ b/amqp/model/snowflake_id.py @@ -2,25 +2,28 @@ import struct class SnowflakeId: + """ + see https://en.wikipedia.org/wiki/Snowflake_ID + """ SNOWFLAKE_ID_WIDTH = 10 # with in bytes (8 bits each), can be more than long int width (8 bytes) - def __init__(self, _input: bytes = None): + def __init__(self, hi_bits_or_bytes = None, lo_bits = None): self.bits = [0, 0] - if _input is not None: - i = 0 - shift = 8 # starting from Hi value, which is only 2 bytes wide - n = 1 - while i < len(_input) and i <= self.SNOWFLAKE_ID_WIDTH: - self.bits[n] |= _input[i] << shift - if shift == 0: - n -= 1 - shift = 56 - else: - shift -= 8 - i += 1 - - def __init__(self, hi_bits: int, lo_bits: int): - self.bits = [lo_bits, hi_bits] + if lo_bits is None: + if hi_bits_or_bytes is not None: + i = 0 + shift = 8 # starting from Hi value, which is only 2 bytes wide + n = 1 + while i < len(hi_bits_or_bytes) and i <= self.SNOWFLAKE_ID_WIDTH: + self.bits[n] |= hi_bits_or_bytes[i] << shift + if shift == 0: + n -= 1 + shift = 56 + else: + shift -= 8 + i += 1 + else: + self.bits = [lo_bits, hi_bits_or_bytes] @staticmethod def from_hex(_input): @@ -42,7 +45,8 @@ class SnowflakeId: def __str__(self): _hex = f"{self.bits[1]:016X}{self.bits[0]:016X}" - return _hex if len(_hex) <= 2 * SnowflakeId.SNOWFLAKE_ID_WIDTH else _hex[(-2 * SnowflakeId.SNOWFLAKE_ID_WIDTH):] + _val = _hex if len(_hex) <= 2 * SnowflakeId.SNOWFLAKE_ID_WIDTH else _hex[(-2 * SnowflakeId.SNOWFLAKE_ID_WIDTH):] + return _val.lower() def __eq__(self, other): if isinstance(other, SnowflakeId): @@ -59,7 +63,7 @@ class SnowflakeId: def get_bytes(self): high = struct.pack("!Q", self.bits[1]) low = struct.pack("!Q", self.bits[0]) - return high[2:] + low + return high[-2:] + low def get_bits(self): return self.bits diff --git a/amqp/model/test_snowflake_id.py b/amqp/model/test_snowflake_id.py new file mode 100644 index 0000000..545cebf --- /dev/null +++ b/amqp/model/test_snowflake_id.py @@ -0,0 +1,48 @@ +from unittest import TestCase + +from amqp.adapter.amq_message_factory import AMQMessageFactory +from amqp.model.snowflake_id import SnowflakeId + + +class SnowflakeIdTest(TestCase): + + def test_equals(self): + _id_1 = AMQMessageFactory.get_instance(1).generate_snowflake_id() + _id_2 = AMQMessageFactory.get_instance(1).generate_snowflake_id() + self.assertFalse(_id_1 == _id_2) + _id_3 = SnowflakeId(_id_1.get_bytes()) + self.assertTrue(_id_1 == _id_3) + + def test_lt(self): + _id_1 = AMQMessageFactory.get_instance(1).generate_snowflake_id() + _id_2 = AMQMessageFactory.get_instance(1).generate_snowflake_id() + self.assertTrue(_id_1 < _id_2) + + + def test_from_hex(self): + _factory = AMQMessageFactory.get_instance(1) + _id_1 = _factory.generate_snowflake_id() + _id_1_ser = _id_1.as_string() + _id_2 = SnowflakeId.from_hex(_id_1_ser) + self.assertTrue(_id_1 == _id_2) + with self.assertRaises(RuntimeError) as context: + SnowflakeId.from_hex("100") + self.assertTrue(f"MessageId is {SnowflakeId.SNOWFLAKE_ID_WIDTH * 8} bits wide, it is represented by HEX string " + in str(context.exception)) + + def test_get_bytes(self): + _factory = AMQMessageFactory.get_instance(1) + _id_1 = _factory.generate_snowflake_id() + _id_1_ser = _id_1.get_bytes() + self.assertEqual(SnowflakeId.SNOWFLAKE_ID_WIDTH, len(_id_1_ser)) + + def test_get_bits(self): + _factory = AMQMessageFactory.get_instance(1) + _id_1 = _factory.generate_snowflake_id() + self.assertEqual(2, len(_id_1.get_bits())) + + + def test_as_string(self): + _factory = AMQMessageFactory.get_instance(1) + _id_1 = _factory.generate_snowflake_id() + self.assertEqual(2 * SnowflakeId.SNOWFLAKE_ID_WIDTH, len(_id_1.as_string())) diff --git a/amqp/rabbitmq/rabbit_mq_consumer.py b/amqp/rabbitmq/rabbit_mq_consumer.py index 44f87a8..7d8880c 100644 --- a/amqp/rabbitmq/rabbit_mq_consumer.py +++ b/amqp/rabbitmq/rabbit_mq_consumer.py @@ -132,7 +132,7 @@ class RabbitMQConsumer(RabbitMQProducer): try: route_queue_name = self.router.findRoute(message) queue_name = route_queue_name.queue if route_queue_name else self.TASK_DISPATCH_QUEUE_NAME - logger.info(f"Get Queue size for {queue_name}") + logging.info(f"Get Queue size for {queue_name}") return max(len(self.awaitingACK), self.channel.message_count(queue_name)) except Exception: return len(self.awaitingACK) diff --git a/amqp/router/route_database.py b/amqp/router/route_database.py index 6a17e30..d7508b1 100644 --- a/amqp/router/route_database.py +++ b/amqp/router/route_database.py @@ -14,14 +14,30 @@ class RouteDatabase: return self.routes def pattern(self, routing_key: str) -> re.Pattern: - regex = r"\.".join( - [ - r".*" if token == "" else r"#" if token == "#" else token - for token in routing_key.split(".") - ] - ) - if routing_key.endswith("."): - regex += r"\." + breaker = False + tokens = routing_key.split('.') + counter = len(tokens) + regex_parts = [] + + for token in tokens: + counter -= 1 + if not breaker: + if token == "": + res = "[^\\.]*" + elif token == "#": + breaker = True + res = ".*$" + else: + res = token + if counter > 0 and not breaker: + res += "\\." + regex_parts.append(res) + else: + regex_parts.append("") + + regex = "".join(filter(lambda x: x != "", regex_parts)) + if not routing_key: + return re.compile(".*") return re.compile(regex) def matches(self, routing_key_from_route: str, routing_key_from_message: str) -> bool: diff --git a/amqp/router/test_route_database.py b/amqp/router/test_route_database.py index 76af304..5b4ac03 100644 --- a/amqp/router/test_route_database.py +++ b/amqp/router/test_route_database.py @@ -1,24 +1,103 @@ from unittest import TestCase +import pytest + +from amqp.adapter.amq_message_factory import AMQMessageFactory +from amqp.adapter.amq_route_factory import AMQRouteFactory +from amqp.model.model import AMQRoute +from amqp.router.route_database import RouteDatabase + +ROUTE_1 = "amq::amq1:clever1.book.localhost.#:5:0" +ROUTE_2 = "amq::amq2:clever2.book.localhost.#:5:0" +ROUTE_3 = "amq::amq3:clever3.book.localhost.#:5:0" +ROUTE_7 = "amq::amq7:clever7.book.localhost.#:5:0" + class TestRouteDatabase(TestCase): + + @pytest.fixture(autouse=True) + def setup_each_test(self): + self._database = RouteDatabase() + if len(self._database.get_routes()) == 0: + r1 = AMQRouteFactory.from_string(ROUTE_1) + r2 = AMQRouteFactory.from_string(ROUTE_2) + r3 = AMQRouteFactory.from_string(ROUTE_3) + self._database.add_route(r1) + self._database.add_route(r2) + self._database.add_route(r3) + def test_get_routes(self): - self.fail() + _amq_factory = AMQRouteFactory() + _route: AMQRoute = _amq_factory.from_string("amq-clever-pybook::clever-pybook:clever.pybook.localhost.#:5:0") + _routes = self._database.get_routes() + self.assertIsNotNone(_routes) + self.assertEqual(3, len(_routes)) + self._database.add_route(_route) + _routes = self._database.get_routes() + self.assertTrue(_route in _routes, "Route was not added") + self.assertEqual(4, len(_routes)) def test_pattern(self): - self.fail() + pattern = self._database.pattern("") + self.assertEqual(".*", pattern.pattern) + pattern = self._database.pattern("..") + self.assertEqual("[^\\.]*\\.[^\\.]*\\.[^\\.]*", pattern.pattern) + pattern = self._database.pattern(".#.#") + self.assertEqual("[^\\.]*\\..*$", pattern.pattern) + pattern = self._database.pattern("..#") + self.assertEqual("[^\\.]*\\.[^\\.]*\\..*$", pattern.pattern) + pattern = self._database.pattern("#") + self.assertEqual(".*$", pattern.pattern) + pattern = self._database.pattern("ba.dc.") + self.assertEqual("ba\\.dc\\.[^\\.]*", pattern.pattern) + pattern = self._database.pattern(".cd.") + self.assertEqual("[^\\.]*\\.cd\\.[^\\.]*", pattern.pattern) + pattern = self._database.pattern("de") + self.assertEqual("de", pattern.pattern) def test_matches(self): - self.fail() + self.assertTrue(self._database.matches("..", "aa.bb.cc")) + self.assertFalse(self._database.matches("..", "aa.bb.")) + self.assertFalse(self._database.matches("..", "aa.bb")) + self.assertFalse(self._database.matches("..", "aa")) + self.assertTrue(self._database.matches("..#", "aa.bb.cc")) + self.assertFalse(self._database.matches("..#", "aa.bb.")) + self.assertFalse(self._database.matches("..#", "aa.bb")) + self.assertFalse(self._database.matches("..#", "aa")) + self.assertTrue(self._database.matches(".#", "aa.bb.cc")) + self.assertTrue(self._database.matches(".#", "aa.bb.")) + self.assertTrue(self._database.matches(".#", "aa.bb")) + self.assertFalse(self._database.matches(".#", "aa")) def test_wrap_routes(self): - self.fail() + wrap = self._database.wrap_routes() + self.assertTrue(ROUTE_1 in wrap) + self.assertTrue(ROUTE_2 in wrap) def test_find_route(self): - self.fail() + message = AMQMessageFactory.get_instance(111).create_request_message( + "POST", "clever3-book.localhost", + "/aa/bb?cc=d", {}, "" + ) + route = self._database.find_route(message.domain, message.path) + self.assertIsNotNone(route) + self.assertEqual(ROUTE_3, route.as_string()) + route = self._database.find_route("clever3.book.localhost", "/three-hot-dogs") + self.assertIsNotNone(route) + self.assertEqual(ROUTE_3, route.as_string()) + route = self._database.find_route("clever-5.book.localhost", "/three-hot-dogs") + self.assertIsNone(route) def test_add_route(self): - self.fail() + origSize = len(self._database.get_routes()) + route = AMQRouteFactory.from_string(ROUTE_7) + self._database.add_route(route) + self.assertEqual(origSize + 1, len(self._database.get_routes())) def test_remove_route(self): - self.fail() + origSize = len(self._database.get_routes()) + route = AMQRouteFactory.from_string(ROUTE_7) + self._database.add_route(route) + self.assertEqual(origSize + 1, len(self._database.get_routes())) + self._database.remove_route(route) + self.assertEqual(origSize, len(self._database.get_routes())) diff --git a/amqp/router/test_router_base.py b/amqp/router/test_router_base.py index 46fd494..c145e07 100644 --- a/amqp/router/test_router_base.py +++ b/amqp/router/test_router_base.py @@ -1,39 +1,122 @@ +from typing import Set from unittest import TestCase +import pytest + +from amqp.adapter.amq_message_factory import AMQMessageFactory +from amqp.adapter.amq_route_factory import AMQRouteFactory +from amqp.model.model import AMQRoute +from amqp.router.router_base import RouterBase +from amqp.router.utils import sanitize_as_rabbitmq_name + +ROUTE_1 = "amq::amq1:clever1.book.localhost.#:5:0" +ROUTE_2 = "amq::amq2:clever2.book.localhost.#:5:0" +ROUTE_3 = "amq::amq3:clever3.book.localhost.#:5:0" +ROUTE_6 = "amq::amq6:clever6.book.localhost.#:5:0" +ROUTE_7 = "amq::amq7:clever7.book.localhost.#:5:0" +ROUTE_8 = "amq::amq8:clever8.book.localhost.#:5:0" +ROUTE_9 = "amq::amq9:clever9.book.localhost.#:5:0" + class TestRouterBase(TestCase): + + @pytest.fixture(autouse=True) + def setup_each_test(self): + self.base = RouterBase() + self.factory = AMQMessageFactory.get_instance(111) + if len(self.base.get_routes()) == 0: + r1 = AMQRouteFactory.from_string(ROUTE_1) + r2 = AMQRouteFactory.from_string(ROUTE_2) + r3 = AMQRouteFactory.from_string(ROUTE_3) + self.base.add_route(r1) + self.base.add_route(r2) + self.base.add_route(r3) + + def test_sanitize_as_rabbitmq_name(self): + self.assertEqual("aa.bb.cc", sanitize_as_rabbitmq_name("aa.bb/cc")) + self.assertEqual("aa.bb.cc", sanitize_as_rabbitmq_name("aa-bb/cc")) + self.assertEqual("aa", sanitize_as_rabbitmq_name("aa&bb/cc")) + self.assertEqual("aa", sanitize_as_rabbitmq_name("aa+bb/cc")) + self.assertEqual("aa", sanitize_as_rabbitmq_name("aa%bb/cc")) + self.assertEqual("aa#bb.cc", sanitize_as_rabbitmq_name("aa#bb/cc")) + self.assertEqual("aa#bb.cc.#", sanitize_as_rabbitmq_name("aa#bb/cc.#")) + self.assertEqual("aa", sanitize_as_rabbitmq_name("aa?bb////cc")) + self.assertEqual("aa.bb.cc", sanitize_as_rabbitmq_name("aa///bb////cc")) + def test_get_routing_key(self): - self.fail() + message = self.factory.create_request_message("POST", "test.me.localhost", "/aa/bb?cc=d", {}, "") + rk = self.base.get_routing_key(message) + self.assertEqual("test.me.localhost.aa.bb", rk) def test_wrap_routes(self): - self.fail() + wrap = self.base.wrap_routes() + self.assertTrue(ROUTE_1 in wrap) + self.assertTrue(ROUTE_2 in wrap) def test_wrap_own_routes(self): - self.fail() + _route: AMQRoute = AMQRouteFactory.from_string(ROUTE_7) + self.base.add_own_route(_route) + o_route = self.base.wrap_own_routes() + self.assertEqual(ROUTE_7, o_route) def test_unwrap_route_list(self): - self.fail() + _routes = self.base.get_routes() + wrapped = self.base.wrap_routes() + unwrapped = self.base.unwrap_route_list(wrapped) + self.assertEqual(len(_routes), len(unwrapped)) + self.assertTrue(unwrapped.pop() in _routes) + self.assertTrue(unwrapped.pop() in _routes) def test_get_routes(self): - self.fail() + _routes = self.base.get_routes() + self.assertEqual(3, len(_routes)) def test_find_route(self): - self.fail() + _route = self.base.find_route("clever3.book.localhost", "/three-hot-dogs") + self.assertIsNotNone(_route) + self.assertEqual(_route.as_string(), ROUTE_3) + _route = self.base.find_route("clever-5.book.localhost", "/three-hot-dogs") + self.assertIsNone(_route) def test_find_route_by_message(self): - self.fail() + message = AMQMessageFactory.get_instance(111).create_request_message( + "POST", "clever3-book.localhost", + "/aa/bb?cc=d", {}, "" + ) + _route: AMQRoute = self.base.find_route_by_message(message) + self.assertIsNotNone(_route) + self.assertEqual(_route.as_string(), ROUTE_3) def test_add_routes(self): - self.fail() + init_routes: Set[AMQRoute] = self.base.get_routes().copy() + self.base.add_routes(ROUTE_6 + ",amq-dlq:DLX:DeadLetterQueue:#:0:0,:::::,:::") + modified = self.base.get_routes() + self.assertEqual(len(init_routes)+1, len(modified)) + wr = self.base.wrap_routes() + self.assertFalse("DLX" in wr) def test_remove_routes(self): - self.fail() + init_routes: Set[AMQRoute] = self.base.get_routes().copy() + toRemove: AMQRoute = init_routes.pop() + self.base.remove_routes(toRemove.as_string()) + modified = self.base.get_routes() + self.assertEqual(len(init_routes), len(modified)) def test_remove_route(self): - self.fail() + origSize = len(self.base.get_routes()) + _route = AMQRouteFactory.from_string(ROUTE_7) + self.base.add_route(_route) + self.assertEqual(origSize + 1, len(self.base.get_routes())) + self.base.remove_route(_route) + self.assertEqual(origSize, len(self.base.get_routes())) def test_add_route(self): - self.fail() + origSize = len(self.base.get_routes()) + _route = AMQRouteFactory.from_string(ROUTE_7) + self.base.add_route(_route) + self.assertEqual(origSize + 1, len(self.base.get_routes())) def test_add_own_route(self): - self.fail() + _route = AMQRouteFactory.from_string(ROUTE_8) + self.base.add_own_route(_route) + self.assertTrue(ROUTE_8 in self.base.wrap_own_routes()) diff --git a/amqp/router/utils.py b/amqp/router/utils.py index 9b18a68..fc15654 100644 --- a/amqp/router/utils.py +++ b/amqp/router/utils.py @@ -1,9 +1,19 @@ import re +pattern = re.compile(r"[^a-zA-Z0-9.#/\-]") + def sanitize_as_rabbitmq_name(input_str: str) -> str: - pattern = re.compile(r"[^a-zA-Z0-9.#/\-]") matcher = pattern.search(input_str) token = input_str[:matcher.start()] if matcher else input_str - step1 = re.sub(r"[^A-Za-z0-9#]", ".", token) - return re.sub(r"\.+", ".", step1) + return sanitize_as_rabbitmq_domain_name(token) + + +def sanitize_as_rabbitmq_domain_name(token: str) -> str: + # Step 1: Replace non-alphanumeric characters (except periods) with a period + step1 = re.sub(r'[^A-Za-z0-9.#]', '.', token) + # Step 2: Replace multiple consecutive periods with a single period + step2 = re.sub(r'\.+', '.', step1) + # Step 3: Remove trailing '.' character, if applicable + step_last_char = len(step2) - 1 + return step2[:-1] if step_last_char > 0 and step2[step_last_char] == '.' else step2 diff --git a/amqp/service/AMQ_PyAdapter.log b/amqp/service/AMQ_PyAdapter.log new file mode 100644 index 0000000..e69de29 diff --git a/amqp/service/amq_service.py b/amqp/service/amq_service.py index 07b4bde..16d106b 100644 --- a/amqp/service/amq_service.py +++ b/amqp/service/amq_service.py @@ -19,7 +19,13 @@ from amqp.service.tracing import initialize_trace logging.basicConfig(level=logging.DEBUG) logging.info(f"CWD = {os.getcwd()}") -logging.config.fileConfig('amqp/service/logging.conf') +logging_conf_path = os.path.join(os.getcwd(), 'amqp', 'service', 'logging.conf') +if os.path.exists(logging_conf_path): + logging.config.fileConfig(logging_conf_path) +else: + logging_conf_path = os.path.join(os.getcwd(), 'logging.conf') + if os.path.exists(logging_conf_path): + logging.config.fileConfig(logging_conf_path) class AMQService: CLEVER_AMQP_CLIENT_VERSION = os.getenv("CLEVER_AMQP_CLIENT_VERSION") diff --git a/amqp/service/service_message_handler.py b/amqp/service/service_message_handler.py index 360866d..bf3e56e 100644 --- a/amqp/service/service_message_handler.py +++ b/amqp/service/service_message_handler.py @@ -82,7 +82,8 @@ class AMQServiceMessageHandler: # map_as_string(amq_message.trace_info), context_with_span, context) if amq_message.magic == AMQMessageFactory.MAGIC_BYTE_HTTP_REQUEST: - response: AMQResponse = self.http_adapter.on_message(amq_message) + response: AMQResponse = asyncio.run(self.http_adapter.on_message(amq_message)) + self.send_reply(properties.reply_to, delivery_tag, channel, response) else: logging.error(" [E] Unknown MAGIC value: [%s], don't know how to interpret the message", amq_message.magic) @@ -137,7 +138,7 @@ class AMQServiceMessageHandler: delivery_tag, reply_to, response.id) reply_props = pika.BasicProperties( correlation_id=response.id, - content_type=response.content_type + content_type=response.content_type[0] ) channel.basic_publish( exchange=AMQ_REPLY_TO_EXCHANGE, diff --git a/core/auth.py b/core/auth.py new file mode 100644 index 0000000..8c5dba1 --- /dev/null +++ b/core/auth.py @@ -0,0 +1,58 @@ +from pytz import timezone + +from typing import Optional +from datetime import datetime, timedelta + +from fastapi.security import OAuth2PasswordBearer + +import jwt + +from core.configs import settings +from core.security import validate_pass, create_password_hash +from schemas.user_schema import UserSchema + +oauth2_schema = OAuth2PasswordBearer( + tokenUrl=f"{settings.API_V0_STR}/login" +) + + +async def authenticate(username: str, password: str) -> Optional[UserSchema]: + if username != 'user': + return None + + test_hash = create_password_hash('user1234') + + if not validate_pass(password, test_hash): + return None + + # The returned value at this point should be UserModel, but since we do not have any persistence + # and this is for demo purposes only + return UserSchema(id=0, username='user', hash_pass=test_hash) + +def _create_token(token_type: str, lifetime: timedelta, sub: str) -> str: + # https://dataracker.ietf.org/doc/html/rfc7519#section-4.1.3 + payload = {} + + pt = timezone('UTC') + expires = datetime.now(tz=pt) + lifetime + + payload["type"] = token_type + + payload["exp"] = expires + + payload["iat"] = datetime.now(tz=pt) #Issued-At (iat) + + payload["sub"] = str(sub) + + return jwt.encode(payload, settings.JWT_SECRET, algorithm=settings.ALGORITHM) + + +def create_access_token(sub: str) -> str: + """ + https://jwt.io + """ + return _create_token( + token_type='access_token', + lifetime=timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES), + sub=sub + ) diff --git a/core/configs.py b/core/configs.py new file mode 100644 index 0000000..e25536a --- /dev/null +++ b/core/configs.py @@ -0,0 +1,28 @@ +from pydantic_settings import BaseSettings +from pathlib import Path + + +class Settings(BaseSettings): + API_V0_STR: str = '/api/v0' + + # JSON Web Tokens (https://jwt.io) - Secret utilized for token generation + JWT_SECRET: str = 'qS96E1oCfq5gEZH-ngD91NC2qkcl0cffhNTIDGpF4pw' + # Token was generated by the code below: + """ + import secrets + + token: str = secrets.token_urlsafe(32) # com 32 caracteres + """ + ALGORITHM: str = 'HS256' + # Token timeout with 1h + ACCESS_TOKEN_EXPIRE_MINUTES: int = 60 + + JOBS_FILE_PATH: Path = 'jobs' + + BASE_PATH: Path = '.' + + class Config: + case_sensitive = True + +settings: Settings = Settings() + diff --git a/core/deps.py b/core/deps.py new file mode 100644 index 0000000..9b32dad --- /dev/null +++ b/core/deps.py @@ -0,0 +1,61 @@ +from typing import Optional + +from fastapi import Depends, HTTPException, status +from fastapi.security import HTTPBasic, HTTPBasicCredentials + +#from jose import jwt, JWTError + +from pydantic import BaseModel + +from schemas.user_schema import UserSchema +from core.auth import oauth2_schema, authenticate +#from core.configs import settings +from core.security import create_password_hash + +class TokenData(BaseModel): + userid: Optional[str] = None + +async def get_current_user( + token: str = Depends(oauth2_schema) +) -> UserSchema: + credential_exception: HTTPException = HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail='User provided credentials are not correct.', + headers={"WWW-Authenticate": "Bearer"} + ) + + try: + #payload = jwt.decode( + # token, + # settings.JWT_SECRET, + # algorithms=[settings.ALGORITHM], + # options={"verify_aud": False} + #) + userid: str = 'user' #payload.get("sub") + if userid is None: + raise credential_exception + + token_data: TokenData = TokenData(userid=userid) + except Exception: + raise credential_exception + + if int(token_data.userid) != 0: + raise credential_exception + + # The returned value at this point should be UserModel, but since we do not have any persistence + # and this is for demo purposes only + return UserSchema(id=token_data.userid, username='user', hash_pass=create_password_hash('user1234')) + + +security = HTTPBasic() + + +async def get_current_username(credentials: HTTPBasicCredentials = Depends(security)) -> str: + user = await authenticate(credentials.username, credentials.password) + if user is None: + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Incorrect username or password", + headers={"WWW-Authenticate": "Basic"}, + ) + return credentials.username diff --git a/core/security.py b/core/security.py new file mode 100644 index 0000000..7f19214 --- /dev/null +++ b/core/security.py @@ -0,0 +1,20 @@ +from passlib.context import CryptContext + +CRIPTO = CryptContext(schemes=['bcrypt'], deprecated='auto') # Automatically workaround deprecated APIs + + +def validate_pass(password: str, hash_pass: str) -> bool: + """ + Verifies if a given password matches the expected hash. + It compares the user provided text password with the stored hash of the expected password, that + is currently stored in the system. + """ + + return CRIPTO.verify(password, hash_pass) + + +def create_password_hash(password: str) -> str: + """ + Return the hash of the password + """ + return CRIPTO.hash(password) \ No newline at end of file diff --git a/requirements.txt b/requirements.txt index 9ea4dd5..fbc9b06 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,9 +1,10 @@ aiohttp~=3.11.11 +fastapi==0.115.6 pika~=1.3.2 opentelemetry-api opentelemetry-sdk opentelemetry-exporter-otlp opentelemetry-exporter-prometheus -opentelemetry-instrumentation-fastapi -opentelemetry-instrumentation-flask opentelemetry-instrumentation-pika +coverage +pytest diff --git a/rest/v0/endpoints/benchmark.py b/rest/v0/endpoints/benchmark.py new file mode 100644 index 0000000..6a369da --- /dev/null +++ b/rest/v0/endpoints/benchmark.py @@ -0,0 +1,111 @@ +import logging + +from fastapi import APIRouter, status, Depends, HTTPException +from fastapi import Path as FPath +from fastapi import Query +from fastapi import UploadFile +from fastapi.responses import FileResponse, Response + +from schemas.request_schema import RequestSchema +from schemas.user_schema import UserSchema +from schemas.job_id_schema import JobIdSchema + +from core.deps import get_current_user + +logger = logging.getLogger('Benchmark') +# Enable propagation to the root logger +logger.propagate = True + +router = APIRouter() + + +# POST Convert +@router.post('/', + status_code=status.HTTP_201_CREATED, summary='Creates a new benchmark job', + description='Creates a new benchmark job/request, however if additional input data files ' + 'are required they must be provided through an addition PUT method call. ' + 'When all the job files have been uploaded a PUT method call must be invoked to' + 'set the job/request status to ReadyForProcessing.', + responses={ + 201: {'model': JobIdSchema, + 'description': 'Job updated with new input files'}, + 507: {'description': 'The server possibly ran out of disk space or an internal server error occurred'} + }) +async def create_benchmark_job(unstructured: UploadFile, + ontology: UploadFile, + ground_truth: UploadFile, + ready_for_processing: bool = Query( + title='Defines if the job is ready for processing', + description='If set to true the job will be marked as ready ' + 'to start being processed. No more files can uploaded ' + 'for this benchmark job, after being marked as ready', + examples=[True, False]), + authenticated_user: UserSchema = Depends(get_current_user)): + logging.info('===> POST /benchmark: Started Benchmark job creation') + logging.info('===> POST arguments: unstructured=%s, ontology=%s, ground_truth=%s, ready_for_processing=%s', + unstructured, ontology, ground_truth, ready_for_processing) + return JobIdSchema(job_id='123456-TestKG') + + +@router.put('/{job_id}', status_code=status.HTTP_202_ACCEPTED, response_model=RequestSchema, + summary='Append additional files to the benchmark request.', + description='Append additional files to the benchmark request. Only possible if Job/Request ' + 'is still in the Created state.', + responses={ + 202: {'model': RequestSchema, + 'description': 'Job updated with new input files'}, + 400: {'description': 'Job does not correspond to a Benchmark'}, + 404: {'description': 'Job ID not found or invalid index'}, + 412: {'description': 'Job is in state that does not allow it to be updated.'}, + 500: {'description': 'A server error occurred while processing the request'} + }) +async def update_benchmark_job(unstructured: UploadFile, + ontology: UploadFile, + ground_truth: UploadFile, + job_id: str = FPath( + title='The Job/Request identifier', + description='The identifier string for the desired Job/Request ' + 'to be updated with additional files for processing', + example='006ea377-baf1-4e04-b18b-37781578d81a'), + ready_for_processing: bool = Query( + title='Defines if the job is ready for processing', + description='If set to true the job will be marked as ready ' + 'to start being processed. No more files can uploaded ' + 'for this benchmark job, after being marked as ready', + examples=[True, False]), + authenticated_user: UserSchema = Depends(get_current_user)): + + logging.info(f'===> PUT /benchmark/{job_id}: Append additional files to the benchmark request') + logging.info('===> PUT arguments: unstructured=%s, ontology=%s, ground_truth=%s, ready_for_processing=%s', + unstructured, ontology, ground_truth, ready_for_processing) + return RequestSchema(job_id=job_id, status='Updated') + +class JSONLResponse(Response): + media_type = "application/jsonl" + + +@router.get('/{request_id}', summary='Polls or Retrieves a benchmark KG by id and index', + description='Polls or Retrieves a processed benchmark KG by id and index. ' + 'If the KG is not ready yet it sends a NO CONTENT (204) HTTP status.', + response_class=JSONLResponse, + response_description='Response is in the form of a jsonl file', + responses={ + 200: {'content': {'c': {}}, + 'description': 'The benchmark KG extracted data'}, + 204: {'description': 'Job is not completed yet'}, + 404: {'description': 'Job ID not found or invalid index'}, + 412: {'description': 'Job does not correspond to a Benchmark'}, + 500: {'description': 'A server error occurred while processing the request'} + }) +async def get_benchmark_kg(request_id: str = + FPath(title='The request id', + description='The id of the request for which the results are to be retrieved', + example='3286f418-8141-450f-8ad5-f6a5c01a0ce9'), + index: int = Query(title='The index of the result to be retrieved', + description='The index of the result to retrieved starting at 0 and ' + 'up to the number of benchmark files -1.', + ge=0, + example=0), + authenticated_user: UserSchema = Depends(get_current_user)): + logging.info(f'===> GET /benchmark/{request_id}?index={index}: Polls or Retrieves a benchmark KG by id and index') + return JSONLResponse(content=f'{"index": {index}, "request_id": "{request_id}", "data": "data"}') diff --git a/rest/v0/endpoints/jobs.py b/rest/v0/endpoints/jobs.py new file mode 100644 index 0000000..8903e22 --- /dev/null +++ b/rest/v0/endpoints/jobs.py @@ -0,0 +1,74 @@ +from typing import List, Optional, Annotated +import logging + +from fastapi import APIRouter, status, Depends, HTTPException +from fastapi import Path as FPath + +from schemas.request_schema import RequestSchema +from schemas.user_schema import UserSchema + +from core.deps import get_current_user + +logger = logging.getLogger('Jobs') +# Enable propagation to the root logger +logger.propagate = True + +router = APIRouter() + +@router.get('/{request_id}', summary='Obtains the status of a given job', + description='Obtains the status of an existing job by its id', + status_code=status.HTTP_200_OK, + responses={ + 200: {'description': 'Job status retrieved successfully'}, + 404: {'description': 'Job not found'}, + 500: {'description': 'A server error occurred while processing the request'} + }) +async def get_job_status(request_id: str = FPath(title='Job ID', + description='The id of the job for which the status is to be queried'), + authenticated_user: UserSchema = Depends(get_current_user)): + logging.info(f'===> GET /jobs/{request_id}: Obtains the status of a given job') + return "JobStatus.Completed" + +@router.get('/', summary='List all user submitted requests', + description='List all user submitted requests along with their status', + status_code=status.HTTP_200_OK, + response_model=List[RequestSchema], + response_description='User submitted job requests', + responses={ + 200: {'description': 'User submitted job requests'}, + 500: {'description': 'A server error occurred while processing the request'} + }) +async def get_requests(authenticated_user: UserSchema = Depends(get_current_user)): + logging.info('===> GET /jobs: List all user submitted requests') + return [] + + +@router.delete('/{request_id}', summary='Deletes a completed or failed job', + description='Deletes a completed or failed job by its request_id', + status_code=status.HTTP_204_NO_CONTENT, + responses={ + 204: {'description': 'Job data was successfully deleted'}, + 404: {'description': 'Job ID not found'}, + 412: {'description': 'Job is in state that does not allow deletion'}, + 500: {'description': 'A server error occurred while processing the request'} + }) +async def delete_job(request_id: str = FPath(title='Request ID', + description='The id of the failed or completed request to be deleted'), + authenticated_user: UserSchema = Depends(get_current_user)): + logging.info(f'===> DELETE /jobs/{request_id}: Deletes a completed or failed job') + return 204 + +@router.put('/{request_id}', summary='Retries execution of a given job', + description='Retries the execution of a given job by its id, provided the current job status is failed', + status_code=status.HTTP_202_ACCEPTED, + responses={ + 202: {'model': RequestSchema, + 'description': 'The updated job status and its configuration'}, + 406: {'description': 'Job is an state that does not allow it to be retried'}, + 500: {'description': 'A server error occurred while processing the request'} + }) +async def retry_job(request_id: str = FPath(title='Job ID', + description='The id of the job to be retried'), + authenticated_user: UserSchema = Depends(get_current_user)): + logging.info(f'===> PUT /jobs/{request_id}: Retries execution of a given job') + return RequestSchema(job_id=request_id, status='Failed') \ No newline at end of file diff --git a/rest/v0/endpoints/unstructured.py b/rest/v0/endpoints/unstructured.py new file mode 100644 index 0000000..11a8e1e --- /dev/null +++ b/rest/v0/endpoints/unstructured.py @@ -0,0 +1,77 @@ +import logging + +from fastapi import APIRouter, status, Depends +from fastapi import Path as FPath +from fastapi import UploadFile + +from schemas.user_schema import UserSchema +from schemas.job_id_schema import JobIdSchema + +from core.deps import get_current_user + +logger = logging.getLogger('Unstructured') +# Enable propagation to the root logger +logger.propagate = True + +router = APIRouter() + + +# POST Convert +@router.post('/with_ontology', status_code=status.HTTP_201_CREATED, summary='Convert unstructured text to KG', + description='Converts unstructured text to a Knowledge Graph according to the provided ontology', + response_description='Returns the Job/Request ID for reference in future interactions', + responses={ + 200: {'model': JobIdSchema, + 'description': 'The KG extraction job ID'}, + 400: {'description': 'Input parameters are not compliant with request'}, + 500: {'description': 'A server error occurred while processing the request'} + }) +async def create_extract_with_ontology(unstructured: UploadFile, + ontology: UploadFile, + ontology_spec: UploadFile, + authenticated_user: UserSchema = Depends(get_current_user)): + logger.debug('====> POST /unstructured/with_ontology: Started convertToKG') + logger.debug('====> arguments: unstructured=%s, ontology=%s, ontology_spec=%s', unstructured, ontology, ontology_spec) + return JobIdSchema(job_id='123456-TestKG') + + +# POST Convert +@router.post('/with_wildcards', status_code=status.HTTP_201_CREATED, + summary='Convert unstructured text to KG with Wildcards filtering', + description='Converts unstructured text to a Knowledge Graph according to the provided ontology, while' + 'applying a filtering based on the Wildcards ontology', + response_description='Knowledge Graph defined by RDF triplets in JSON format', + responses={ + 200: {'model': JobIdSchema, + 'description': 'The KG extraction job ID'}, + 400: {'description': 'Input parameters are not compliant with request'}, + 500: {'description': 'A server error occurred while processing the request'} + }) +async def create_extract_with_ontology(unstructured: UploadFile, + ontology: UploadFile, + ontology_spec: UploadFile, + wildcards: UploadFile, + authenticated_user: UserSchema = Depends(get_current_user)): + logger.debug('====> POST /unstructured/with_wildcards: Started convertToKG with Wildcards filtering') + logger.debug('====> arguments: unstructured=%s, ontology=%s, ontology_spec=%s, wildcards=%s', + unstructured, ontology, ontology_spec, wildcards) + return JobIdSchema(job_id='123456-TestKG') + + +@router.get('/{request_id}', summary='Polls or Retrieves a KG by id', + description='Polls or Retrieves a processed KG by id. If the KG is not ready yet it sends a ' + 'NO CONTENT (204) HTTP status', + response_description='Response is in the form of json data', + responses={ + 200: {'description': 'The KG extracted data'}, + 204: {'description': 'Job is not completed yet'}, + 404: {'description': 'Job ID not found'}, + 412: {'description': 'Job does not correspond to a KG extraction job'}, + 500: {'description': 'A server error occurred while processing the request'} + }) +async def get_kg(request_id: str = FPath(title='The job id', + description='The id of the job for which the results are to be retrieved'), + authenticated_user: UserSchema = Depends(get_current_user)): + logger.debug(f'====> GET /unstructured/{request_id}: Started convertToKG with Wildcards filtering') + + diff --git a/schemas/job_id_schema.py b/schemas/job_id_schema.py new file mode 100644 index 0000000..f3fe4ad --- /dev/null +++ b/schemas/job_id_schema.py @@ -0,0 +1,5 @@ +from pydantic import BaseModel + + +class JobIdSchema(BaseModel): + job_id: str \ No newline at end of file diff --git a/schemas/request_schema.py b/schemas/request_schema.py new file mode 100644 index 0000000..bc8119d --- /dev/null +++ b/schemas/request_schema.py @@ -0,0 +1,17 @@ +from typing import Optional, List +from datetime import datetime + +from pydantic import BaseModel + + +class RequestSchema(BaseModel): + id: str = None + created: datetime + type: str + status: str + retries_count: int + unstructured_sources: List[str] + ontology_sources: List[str] + ontology_spec_sources: List[str] + wildcards_sources: List[str] + ground_truth_sources: List[str] diff --git a/schemas/user_schema.py b/schemas/user_schema.py new file mode 100644 index 0000000..ccc0c38 --- /dev/null +++ b/schemas/user_schema.py @@ -0,0 +1,9 @@ +from typing import Optional + +from pydantic import BaseModel + + +class UserSchema(BaseModel): + id: Optional[int] = None + username: str + hash_pass: str