clevermicro/amq-adapter-python
14
0
Fork 0
Code Issues 16 Pull Requests 4 Actions Packages Projects Releases Wiki Activity

Remove references to cleverswarm and fix packaging

Browse Source
This commit is contained in:
Stanislav Hejny
2025-02-27 21:43:46 +00:00
parent 30e046b234
commit e861bf0f38
5 changed files with 18 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
amqp/adapter/cleverthis_service_adapter.py
+18 -21
View File
@@ -1,23 +1,20 @@
import json
import logging
import re
from typing import Dict, List, Tuple, Callable, Coroutine, AnyStr, Any
import python_multipart
import io
import re
from aiohttp import ClientSession, ClientResponse
from fastapi import HTTPException, UploadFile
from python_multipart.multipart import Field, File
from typing import Dict, List, Tuple, Callable, Coroutine, AnyStr, Any
from urllib.parse import parse_qs
from xml.etree import ElementTree
from amqp.adapter.amq_response_factory import AMQResponseFactory
from amqp.config.amq_configuration import AMQConfiguration
from amqp.model.model import AMQMessage, AMQResponse
from cleverswarm.core.deps import get_current_user
from cleverswarm.schemas.user_schema import UserSchema
import json
from urllib.parse import parse_qs
from xml.etree import ElementTree
import io
class CleverMultiPartParser:
@@ -96,9 +93,9 @@ def parse_request_body(message: AMQMessage):
async def call_cleverswarm_post_put_api(
message: AMQMessage,
args: Any,
api_method: Callable[[Any, UserSchema], Coroutine],
api_method: Callable[[Any, Any], Coroutine],
success_code: int,
authenticated_user: UserSchema
authenticated_user: Any
) -> AMQResponse:
try:
coroutine_call: Coroutine = api_method(args, authenticated_user)
@@ -119,8 +116,8 @@ async def call_cleverswarm_post_put_api(
async def call_cleverswarm_get_api(
message: AMQMessage,
pattern: str,
api_method: Callable[[Tuple[AnyStr | Any, ...], UserSchema], Coroutine],
authenticated_user: UserSchema
api_method: Callable[[Tuple[AnyStr | Any, ...], Any], Coroutine],
authenticated_user: Any
) -> AMQResponse:
try:
match = re.search(pattern, message.path)
@@ -166,7 +163,7 @@ class CleverThisServiceAdapter:
"""
DEBUG_NO_AUTH = 1
_auth: str = message.headers.get('Authorization', '')
_auth_user: UserSchema | None = None
_auth_user: Any | None = None
if 'Bearer' in _auth:
try:
token = _auth.split(' ')[1]
@@ -194,7 +191,7 @@ class CleverThisServiceAdapter:
# ===================== C l e v e r T h i s A P I m e t h o d s ============================
# ==================================================================================================
async def get(self, message: AMQMessage, auth_user: UserSchema) -> AMQResponse:
async def get(self, message: AMQMessage, auth_user: Any) -> AMQResponse:
"""
Handles the GET requests for the CleverThis API. to be overriden in subclass. This is a base implementation.
It will attempt to call the REST endpoint directly using HTTP protocol.
@@ -215,7 +212,7 @@ class CleverThisServiceAdapter:
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:
async def put(self, message: AMQMessage, auth_user: Any) -> AMQResponse:
"""
Handles the PUT requests for the CleverThis API. to be overriden in subclass. This is a base implementation.
It will attempt to call the REST endpoint directly using HTTP protocol.
@@ -232,7 +229,7 @@ class CleverThisServiceAdapter:
auth_user
)
async def post(self, message: AMQMessage, auth_user: UserSchema) -> AMQResponse:
async def post(self, message: AMQMessage, auth_user: Any) -> AMQResponse:
"""
Handles the POST requests for the CleverThis API. to be overriden in subclass. This is a base implementation.
It will attempt to call the REST endpoint directly using HTTP protocol.
@@ -249,10 +246,10 @@ class CleverThisServiceAdapter:
auth_user
)
async def handle_possible_form(self, request_coroutine, auth_user: UserSchema) -> AMQResponse:
async def handle_possible_form(self, request_coroutine, auth_user: Any) -> AMQResponse:
return await request_coroutine
async def head(self, message: AMQMessage, auth_user: UserSchema) -> AMQResponse:
async def head(self, message: AMQMessage, auth_user: Any) -> AMQResponse:
"""
Handles the HEAD requests for the CleverThis API. to be overriden in subclass. This is a base implementation.
It will attempt to call the REST endpoint directly using HTTP protocol.
@@ -262,7 +259,7 @@ class CleverThisServiceAdapter:
"""
raise NotImplementedError
async def delete(self, message: AMQMessage, auth_user: UserSchema) -> AMQResponse:
async def delete(self, message: AMQMessage, auth_user: Any) -> AMQResponse:
"""
Handles the DELETE requests for the CleverThis API. to be overriden in subclass. This is a base implementation.
It will attempt to call the REST endpoint directly using HTTP protocol.