Movatterモバイル変換

[0]ホーム
URL:

Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly

 Sign up
Appearance settings

Add sqs operations#6

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to ourterms of service andprivacy statement. We’ll occasionally send you account related emails.

Already on GitHub?Sign in to your account

Merged
giograno merged 8 commits intomainfromsqs
Oct 23, 2024
Merged
Show file tree
Hide file tree
Changes fromall commits
Commits
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions.github/workflows/test.yml
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -4,9 +4,9 @@ on:
pull_request:
paths:
- .github/workflows/test.yml
- localstack-sdk-python/*
- tests/*
- packages/*
- localstack-sdk-python/**
- tests/**
- packages/**
push:
branches:
- main
Expand Down
3 changes: 3 additions & 0 deletionslocalstack-sdk-python/localstack/sdk/aws/__init__.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,3 @@
from localstack.sdk.aws.client import AWSClient

__all__ = ["AWSClient"]
51 changes: 51 additions & 0 deletionslocalstack-sdk-python/localstack/sdk/aws/client.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,51 @@
import json

from localstack.clients import BaseClient
from localstack.sdk.api.aws_api import AwsApi
from localstack.sdk.models import Message


def _from_sqs_query_to_json(xml_dict: dict) -> list[Message]:
"""
todo: developer endpoint implements sqs-query protocol. Remove this workaround one we move them to json.
"""
raw_messages = (
xml_dict.get("ReceiveMessageResponse", {}).get("ReceiveMessageResult", {}) or {}
).get("Message", [])
if isinstance(raw_messages, dict):
raw_messages = [raw_messages]
messages = []
for msg in raw_messages:
_attributes = msg.get("Attribute", [])
attributes = {i["Name"]: i["Value"] for i in _attributes}
_m = {
"MessageId": msg.get("MessageId"),
"ReceiptHandle": msg.get("ReceiptHandle"),
"MD5OfBody": msg.get("MD5OfBody"),
"Body": msg.get("Body"),
"Attributes": attributes,
}
m = Message.from_dict(_m)
messages.append(m)
return messages


class AWSClient(BaseClient):
def __init__(self, **kwargs) -> None:
super().__init__(**kwargs)
self._client = AwsApi(self._api_client)

def list_sqs_messages(self, account_id: str, region: str, queue_name: str) -> list[Message]:
response = self._client.list_sqs_messages_with_http_info(
account_id=account_id, region=region, queue_name=queue_name
)
return _from_sqs_query_to_json(json.loads(response.raw_data))

def list_sqs_messages_from_queue_url(self, queue_url) -> list[Message]:
response = self._client.list_all_sqs_messages_with_http_info(queue_url=queue_url)
return _from_sqs_query_to_json(json.loads(response.raw_data))


def get_default(**args) -> AwsApi:
"""Return a client with a default configuration"""
return AwsApi(args)
Loading
[8]ページ先頭
©2009-2025 Movatter.jp