ministack-aws-emulator

仓库: aradotso/trending-skills
安装量: 294
排名: #6919

安装

npx skills add https://github.com/aradotso/trending-skills --skill ministack-aws-emulator

MiniStack AWS Emulator Skill by ara.so — Daily 2026 Skills collection. MiniStack is a free, MIT-licensed drop-in replacement for LocalStack that emulates 25+ AWS services (S3, SQS, DynamoDB, Lambda, SNS, IAM, STS, Kinesis, EventBridge, SecretsManager, SSM, CloudWatch, SES, and more) on a single port ( 4566 ). No account, no API key, no telemetry. Works with boto3 , AWS CLI, Terraform, CDK, and any SDK. Installation Option 1: PyPI (simplest) pip install ministack ministack

Server runs at http://localhost:4566

Change port: GATEWAY_PORT=5000 ministack

Option 2: Docker Hub docker run -p 4566 :4566 nahuelnucera/ministack Option 3: Docker Compose (from source) git clone https://github.com/Nahuel990/ministack cd ministack docker compose up -d Verify it's running curl http://localhost:4566/_localstack/health Configuration Environment Variable Default Description GATEWAY_PORT 4566 Port to listen on S3_PERSIST 0 Set to 1 to persist S3 data to disk AWS CLI Usage

Set credentials (any non-empty values work)

export AWS_ACCESS_KEY_ID = test export AWS_SECRET_ACCESS_KEY = test export AWS_DEFAULT_REGION = us-east-1

S3

aws --endpoint-url

http://localhost:4566 s3 mb s3://my-bucket aws --endpoint-url = http://localhost:4566 s3 cp ./file.txt s3://my-bucket/ aws --endpoint-url = http://localhost:4566 s3 ls s3://my-bucket

SQS

aws --endpoint-url

http://localhost:4566 sqs create-queue --queue-name my-queue aws --endpoint-url = http://localhost:4566 sqs list-queues

DynamoDB

aws --endpoint-url

http://localhost:4566 dynamodb list-tables aws --endpoint-url = http://localhost:4566 dynamodb create-table \ --table-name Users \ --attribute-definitions AttributeName = userId,AttributeType = S \ --key-schema AttributeName = userId,KeyType = HASH \ --billing-mode PAY_PER_REQUEST

STS (identity check)

aws --endpoint-url

http://localhost:4566 sts get-caller-identity

Use a named profile instead

aws configure --profile local

Enter: test / test / us-east-1 / json

aws --profile local --endpoint-url = http://localhost:4566 s3 ls awslocal wrapper (from source) chmod +x bin/awslocal ./bin/awslocal s3 ls ./bin/awslocal dynamodb list-tables boto3 Usage Patterns Universal client factory import boto3 ENDPOINT = "http://localhost:4566" def aws_client ( service : str ) : return boto3 . client ( service , endpoint_url = ENDPOINT , aws_access_key_id = "test" , aws_secret_access_key = "test" , region_name = "us-east-1" , ) def aws_resource ( service : str ) : return boto3 . resource ( service , endpoint_url = ENDPOINT , aws_access_key_id = "test" , aws_secret_access_key = "test" , region_name = "us-east-1" , ) S3 s3 = aws_client ( "s3" )

Create bucket and upload

s3 . create_bucket ( Bucket = "my-bucket" ) s3 . put_object ( Bucket = "my-bucket" , Key = "hello.txt" , Body = b"Hello, MiniStack!" )

Download

obj

s3 . get_object ( Bucket = "my-bucket" , Key = "hello.txt" ) print ( obj [ "Body" ] . read ( ) )

b'Hello, MiniStack!'

List objects

response

s3 . list_objects_v2 ( Bucket = "my-bucket" ) for item in response . get ( "Contents" , [ ] ) : print ( item [ "Key" ] )

Copy object

s3 . copy_object ( Bucket = "my-bucket" , CopySource = { "Bucket" : "my-bucket" , "Key" : "hello.txt" } , Key = "hello-copy.txt" , )

Enable versioning

s3 . put_bucket_versioning ( Bucket = "my-bucket" , VersioningConfiguration = { "Status" : "Enabled" } , )

Presigned URL (works locally)

url

s3 . generate_presigned_url ( "get_object" , Params = { "Bucket" : "my-bucket" , "Key" : "hello.txt" } , ExpiresIn = 3600 , ) SQS sqs = aws_client ( "sqs" )

Standard queue

queue

sqs . create_queue ( QueueName = "my-queue" ) queue_url = queue [ "QueueUrl" ] sqs . send_message ( QueueUrl = queue_url , MessageBody = '{"event": "user_signup"}' ) messages = sqs . receive_message ( QueueUrl = queue_url , MaxNumberOfMessages = 10 ) for msg in messages . get ( "Messages" , [ ] ) : print ( msg [ "Body" ] ) sqs . delete_message ( QueueUrl = queue_url , ReceiptHandle = msg [ "ReceiptHandle" ] )

FIFO queue

fifo

sqs . create_queue ( QueueName = "my-queue.fifo" , Attributes = { "FifoQueue" : "true" , "ContentBasedDeduplication" : "true" } , )

Dead-letter queue setup

dlq

sqs . create_queue ( QueueName = "my-dlq" ) dlq_attrs = sqs . get_queue_attributes ( QueueUrl = dlq [ "QueueUrl" ] , AttributeNames = [ "QueueArn" ] ) sqs . set_queue_attributes ( QueueUrl = queue_url , Attributes = { "RedrivePolicy" : json . dumps ( { "deadLetterTargetArn" : dlq_attrs [ "Attributes" ] [ "QueueArn" ] , "maxReceiveCount" : "3" , } ) } , ) DynamoDB import json ddb = aws_client ( "dynamodb" )

Create table

ddb . create_table ( TableName = "Users" , KeySchema = [ { "AttributeName" : "userId" , "KeyType" : "HASH" } , { "AttributeName" : "createdAt" , "KeyType" : "RANGE" } , ] , AttributeDefinitions = [ { "AttributeName" : "userId" , "AttributeType" : "S" } , { "AttributeName" : "createdAt" , "AttributeType" : "N" } , ] , BillingMode = "PAY_PER_REQUEST" , )

Put / Get / Delete

ddb . put_item ( TableName = "Users" , Item = { "userId" : { "S" : "u1" } , "createdAt" : { "N" : "1700000000" } , "name" : { "S" : "Alice" } , "active" : { "BOOL" : True } , } , ) item = ddb . get_item ( TableName = "Users" , Key = { "userId" : { "S" : "u1" } , "createdAt" : { "N" : "1700000000" } } , ) print ( item [ "Item" ] [ "name" ] [ "S" ] )

Alice

Query

result

ddb . query ( TableName = "Users" , KeyConditionExpression = "userId = :uid" , ExpressionAttributeValues = { ":uid" : { "S" : "u1" } } , )

Batch write

ddb . batch_write_item ( RequestItems = { "Users" : [ { "PutRequest" : { "Item" : { "userId" : { "S" : "u2" } , "createdAt" : { "N" : "1700000001" } , "name" : { "S" : "Bob" } } } } , ] } )

TTL

ddb . update_time_to_live ( TableName = "Users" , TimeToLiveSpecification = { "Enabled" : True , "AttributeName" : "expiresAt" } , ) SNS + SQS Fanout sns = aws_client ( "sns" ) sqs = aws_client ( "sqs" ) topic = sns . create_topic ( Name = "my-topic" ) topic_arn = topic [ "TopicArn" ] queue = sqs . create_queue ( QueueName = "fan-queue" ) queue_attrs = sqs . get_queue_attributes ( QueueUrl = queue [ "QueueUrl" ] , AttributeNames = [ "QueueArn" ] ) queue_arn = queue_attrs [ "Attributes" ] [ "QueueArn" ] sns . subscribe ( TopicArn = topic_arn , Protocol = "sqs" , Endpoint = queue_arn )

Publish — message is fanned out to subscribed SQS queues

sns . publish ( TopicArn = topic_arn , Message = "hello fanout" , Subject = "test" ) Lambda import zipfile , io

Create a zip with handler code

buf

io . BytesIO ( ) with zipfile . ZipFile ( buf , "w" ) as zf : zf . writestr ( "handler.py" , """ def handler(event, context): print("event:", event) return {"statusCode": 200, "body": "ok"} """ ) buf . seek ( 0 ) lam = aws_client ( "lambda" ) lam . create_function ( FunctionName = "my-function" , Runtime = "python3.12" , Role = "arn:aws:iam::000000000000:role/role" , Handler = "handler.handler" , Code = { "ZipFile" : buf . read ( ) } , )

Invoke synchronously

import json response = lam . invoke ( FunctionName = "my-function" , InvocationType = "RequestResponse" , Payload = json . dumps ( { "key" : "value" } ) , ) result = json . loads ( response [ "Payload" ] . read ( ) ) print ( result )

{"statusCode": 200, "body": "ok"}

SQS event source mapping

lam . create_event_source_mapping ( EventSourceArn = queue_arn , FunctionName = "my-function" , BatchSize = 10 , Enabled = True , ) SecretsManager sm = aws_client ( "secretsmanager" ) sm . create_secret ( Name = "db-password" , SecretString = '{"password":"s3cr3t"}' ) secret = sm . get_secret_value ( SecretId = "db-password" ) print ( secret [ "SecretString" ] )

{"password":"s3cr3t"}

sm . update_secret ( SecretId = "db-password" , SecretString = '{"password":"newpass"}' ) sm . delete_secret ( SecretId = "db-password" , ForceDeleteWithoutRecovery = True ) SSM Parameter Store ssm = aws_client ( "ssm" ) ssm . put_parameter ( Name = "/app/db/host" , Value = "localhost" , Type = "String" ) ssm . put_parameter ( Name = "/app/db/password" , Value = "secret" , Type = "SecureString" ) param = ssm . get_parameter ( Name = "/app/db/host" ) print ( param [ "Parameter" ] [ "Value" ] )

localhost

Fetch all params under a path

params

ssm . get_parameters_by_path ( Path = "/app/" , Recursive = True ) for p in params [ "Parameters" ] : print ( p [ "Name" ] , p [ "Value" ] ) Kinesis import base64 kin = aws_client ( "kinesis" ) kin . create_stream ( StreamName = "events" , ShardCount = 1 ) kin . put_record ( StreamName = "events" , Data = b'{"event":"click"}' , PartitionKey = "user1" )

Get records

shards

kin . list_shards ( StreamName = "events" ) shard_id = shards [ "Shards" ] [ 0 ] [ "ShardId" ] iterator = kin . get_shard_iterator ( StreamName = "events" , ShardId = shard_id , ShardIteratorType = "TRIM_HORIZON" , ) records = kin . get_records ( ShardIterator = iterator [ "ShardIterator" ] ) for r in records [ "Records" ] : print ( base64 . b64decode ( r [ "Data" ] ) ) EventBridge eb = aws_client ( "events" )

Create a custom bus

eb . create_event_bus ( Name = "my-bus" )

Put a rule targeting a Lambda

eb . put_rule ( Name = "my-rule" , EventBusName = "my-bus" , EventPattern = '{"source": ["myapp"]}' , State = "ENABLED" , ) eb . put_targets ( Rule = "my-rule" , EventBusName = "my-bus" , Targets = [ { "Id" : "1" , "Arn" : "arn:aws:lambda:us-east-1:000000000000:function:my-function" } ] , )

Emit an event (triggers Lambda target)

eb . put_events ( Entries = [ { "Source" : "myapp" , "DetailType" : "UserSignup" , "Detail" : '{"userId": "123"}' , "EventBusName" : "my-bus" , } ] ) CloudWatch Logs import time logs = aws_client ( "logs" ) logs . create_log_group ( logGroupName = "/app/service" ) logs . create_log_stream ( logGroupName = "/app/service" , logStreamName = "stream-1" ) logs . put_log_events ( logGroupName = "/app/service" , logStreamName = "stream-1" , logEvents = [ { "timestamp" : int ( time . time ( ) * 1000 ) , "message" : "App started" } , { "timestamp" : int ( time . time ( ) * 1000 ) , "message" : "Request received" } , ] , ) events = logs . get_log_events ( logGroupName = "/app/service" , logStreamName = "stream-1" , ) for e in events [ "events" ] : print ( e [ "message" ] )

Filter with glob patterns (* and ?), AND terms, -exclusions

filtered

logs . filter_log_events ( logGroupName = "/app/service" , filterPattern = "Request*" , ) Testing Patterns pytest fixture (recommended) import pytest import boto3 MINISTACK_ENDPOINT = "http://localhost:4566" @pytest . fixture ( scope = "session" ) def aws_endpoint ( ) : return MINISTACK_ENDPOINT @pytest . fixture def s3_client ( aws_endpoint ) : return boto3 . client ( "s3" , endpoint_url = aws_endpoint , aws_access_key_id = "test" , aws_secret_access_key = "test" , region_name = "us-east-1" , ) @pytest . fixture def test_bucket ( s3_client ) : bucket = "test-bucket" s3_client . create_bucket ( Bucket = bucket ) yield bucket

Cleanup

objs

s3_client . list_objects_v2 ( Bucket = bucket ) . get ( "Contents" , [ ] ) for obj in objs : s3_client . delete_object ( Bucket = bucket , Key = obj [ "Key" ] ) s3_client . delete_bucket ( Bucket = bucket ) def test_upload_download ( s3_client , test_bucket ) : s3_client . put_object ( Bucket = test_bucket , Key = "test.txt" , Body = b"hello" ) resp = s3_client . get_object ( Bucket = test_bucket , Key = "test.txt" ) assert resp [ "Body" ] . read ( ) == b"hello" GitHub Actions CI integration

.github/workflows/test.yml

name : Test on : [ push , pull_request ] jobs : test : runs-on : ubuntu - latest services : ministack : image : nahuelnucera/ministack ports : - 4566 : 4566 steps : - uses : actions/checkout@v4 - uses : actions/setup - python@v5 with : python-version : "3.12" - run : pip install - r requirements.txt - run : pytest env : AWS_ACCESS_KEY_ID : test AWS_SECRET_ACCESS_KEY : test AWS_DEFAULT_REGION : us - east - 1 AWS_ENDPOINT_URL : http : //localhost : 4566 Using AWS_ENDPOINT_URL env var (boto3 >= 1.28) import os import boto3

If AWS_ENDPOINT_URL is set, boto3 uses it automatically — no endpoint_url kwarg needed

export AWS_ENDPOINT_URL=http://localhost:4566

s3

boto3 . client ( "s3" )

picks up AWS_ENDPOINT_URL automatically

Supported Services (25+) Service Key Operations S3 CRUD, multipart, versioning, encryption, lifecycle, CORS, ACL, notifications SQS Standard & FIFO queues, DLQ, batch ops SNS Topics, subscriptions, fanout to SQS/Lambda, platform endpoints DynamoDB Tables, CRUD, Query, Scan, TTL, transactions, batch ops Lambda Python runtimes, invoke, SQS event sources, Function URLs IAM Users, roles, policies, groups, instance profiles, OIDC STS GetCallerIdentity, AssumeRole, GetSessionToken SecretsManager Full CRUD, rotation, versioning SSM Parameter Store String, SecureString, StringList, path queries EventBridge Buses, rules, targets, Lambda dispatch Kinesis Streams, shards, records, iterators CloudWatch Metrics PutMetricData, alarms, dashboards, CBOR protocol CloudWatch Logs Log groups/streams, filter with globs, metric filters SES Send email, templates, configuration sets Step Functions State machine CRUD RDS Spins up real Postgres/MySQL containers ElastiCache Spins up real Redis containers Athena Real SQL via DuckDB ECS Real Docker containers Troubleshooting Connection refused on port 4566

Check if ministack is running

curl http://localhost:4566/_localstack/health

Start it

ministack

or

docker run -p 4566 :4566 nahuelnucera/ministack NoCredentialsError from boto3 export AWS_ACCESS_KEY_ID = test export AWS_SECRET_ACCESS_KEY = test export AWS_DEFAULT_REGION = us-east-1

Any non-empty values work — MiniStack doesn't validate credentials

InvalidSignatureException This is usually a region mismatch. Ensure region_name="us-east-1" matches across all clients. Lambda function not found after create MiniStack executes Python runtimes with a warm worker pool. Wait briefly or invoke with InvocationType="Event" for async. S3 data lost on restart

Enable persistence

S3_PERSIST

1 ministack

or in Docker

docker run -p 4566 :4566 -e S3_PERSIST = 1 -v $( pwd ) /data:/data nahuelnucera/ministack Port conflict GATEWAY_PORT = 5000 ministack

Then use http://localhost:5000 as endpoint

Migrating from LocalStack Replace all http://localhost:4566 endpoint URLs — they stay the same. Remove LOCALSTACK_AUTH_TOKEN / LOCALSTACK_API_KEY env vars (not needed). Replace localstack/localstack Docker image with nahuelnucera/ministack . All boto3 client code works without modification.

← 返回排行榜