🚀 This project demonstrates how to send messages to an AWS SQS queue using FastAPI and LocalStack for local testing.
Amazon Simple Queue Service (SQS) is a fully managed message queuing service that enables decoupling and scaling of microservices, distributed systems, and serverless applications. Some key benefits include:
- Scalability: Easily handle high-throughput workloads.
- Reliability: Built-in redundancy ensures message durability.
- Decoupling: Separates different parts of your application for better maintainability.
- Security: Supports encryption and access control via AWS IAM policies.
- Asynchronous Processing: Ideal for background tasks and event-driven architectures.
.
├── .env
├── main.py
├── core
│ └── config.py
├── dependencies
│ └── aws_sqs.py
├── routes
│ └── api
│ └── messages.py
├── schemas
│ └── messages.py
└── utils
└── logger.py
- Python 3.8+
- uv (for dependency management)
1️⃣ Clone the repository:
git clone https://github.com/egoan82/fastapi-with-aws-sqs.git
cd fastapi-with-aws-sqs
2️⃣ Create and activate a virtual environment with uv
:
uv venv
source .venv/bin/activate
3️⃣ Install dependencies:
pip install -r requirements.txt
4️⃣ Configure environment variables in .env
:
AWS_KEY_ID="test"
AWS_SECRET_KEY="test"
AWS_REGION="us-east-1"
SQS_QUEUE_URL="https://localhost.localstack.cloud:4566/000000000000/queue.fifo"
LOCALSTACK_URL="https://localhost.localstack.cloud:4566"
1️⃣ Run the FastAPI application:
uvicorn main:app --reload
️2️⃣ Run listener to receive messages:
python main.py
3️⃣ Access the interactive API documentation:
🔗 Swagger UI
🔗 ReDoc
1️⃣ Make a POST request to /send_message
with the following body:
{
"title": "string",
"message": "string"
}
2️⃣ The API responds with the message_id
of the sent message.
Here are some screenshots of the API in action:
🚀 Happy coding and enjoy building with FastAPI and AWS SQS! 🎉