Appendix A — Other services
Six services outside the book’s data focus that oblako also runs, each with one call that works on it.
A.1 Before you start
oblako up moto
oblako up stepfunctions
oblako up bedrock-runtime
oblako up ecs
export AWS_PROFILE=oblakoLambda and ECS run containers on Docker, and Bedrock answers from a local model in Ollama, which oblako expects on its default port with qwen2.5:0.5b pulled (oblako pull). The last listing removes everything the others create.
A.2 Lambda
oblako keeps Lambda’s control plane in moto and runs each invocation in a container from a Lambda runtime image, so the handler runs as it would on AWS:
import io
import json
import zipfile
import boto3
HANDLER = """
def handler(event, context):
total = sum(item["amount"] for item in event["orders"])
return {"orders": len(event["orders"]), "total": total}
"""
package = io.BytesIO()
with zipfile.ZipFile(package, "w") as archive:
archive.writestr("app.py", HANDLER)
# Lambda needs a role it may assume, on oblako as on AWS
role = boto3.client("iam").create_role(
RoleName="order-total",
AssumeRolePolicyDocument=json.dumps(
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Principal": {
"Service": "lambda.amazonaws.com"
},
"Action": "sts:AssumeRole",
}
],
}
),
)["Role"]["Arn"]
lam = boto3.client("lambda")
lam.create_function(
FunctionName="order-total",
Runtime="python3.12",
Role=role,
Handler="app.handler",
Code={"ZipFile": package.getvalue()},
)
lam.get_waiter("function_active_v2").wait(
FunctionName="order-total"
)
event = {"orders": [{"amount": 12.5}, {"amount": 7.5}]}
resp = lam.invoke(
FunctionName="order-total", Payload=json.dumps(event)
)
print(resp["StatusCode"], json.loads(resp["Payload"].read()))200 {'orders': 2, 'total': 20.0}As on AWS, the function’s role must exist and trust lambda.amazonaws.com, or CreateFunction fails. The runtime image is x86_64 with Python 3.12, which Docker emulates on an ARM machine; the first invocation also pulls the image.
A.3 Step Functions
oblako runs aws-stepfunctions-local, AWS’s own engine for running Step Functions on one machine, which executes the states of a definition:
import json
import time
import boto3
definition = {
"StartAt": "Check",
"States": {
"Check": {
"Type": "Choice",
"Choices": [
{
"Variable": "$.rows",
"NumericGreaterThan": 0,
"Next": "Load",
}
],
"Default": "Skip",
},
"Load": {"Type": "Pass", "Result": "loaded", "End": True},
"Skip": {
"Type": "Pass",
"Result": "skipped",
"End": True,
},
},
}
sfn = boto3.client("stepfunctions")
machine = sfn.create_state_machine(
name="daily-load",
definition=json.dumps(definition),
roleArn="arn:aws:iam::123456789012:role/sfn",
)["stateMachineArn"]
for rows in (40, 0):
run = sfn.start_execution(
stateMachineArn=machine, input=json.dumps({"rows": rows})
)["executionArn"]
while (desc := sfn.describe_execution(executionArn=run))[
"status"
] == "RUNNING":
time.sleep(0.2)
print(rows, desc["status"], desc["output"])40 SUCCEEDED "loaded"
0 SUCCEEDED "skipped"A Task state that calls a Lambda function needs that function reachable from the engine; Pass, Choice, Wait and the other flow states need nothing more.
A.4 EventBridge
moto delivers matching events to SQS, SNS and Lambda targets; oblako adds delivery to Redshift Data API targets and fires scheduled rules:
import json
import boto3
sqs = boto3.client("sqs")
events = boto3.client("events")
queue = sqs.create_queue(QueueName="file-arrivals")["QueueUrl"]
queue_arn = sqs.get_queue_attributes(
QueueUrl=queue, AttributeNames=["QueueArn"]
)["Attributes"]["QueueArn"]
events.put_rule(
Name="orders-landed",
EventPattern=json.dumps(
{
"source": ["shop.ingest"],
"detail-type": ["File landed"],
}
),
)
events.put_targets(
Rule="orders-landed",
Targets=[{"Id": "queue", "Arn": queue_arn}],
)
events.put_events(
Entries=[
{
"Source": "shop.ingest",
"DetailType": "File landed",
"Detail": json.dumps(
{"key": "orders/date=2026-09-30/"}
),
}
]
)
msg = sqs.receive_message(QueueUrl=queue, WaitTimeSeconds=5)
body = json.loads(msg["Messages"][0]["Body"])
print(body["detail-type"], body["detail"])File landed {'key': 'orders/date=2026-09-30/'}A.5 Bedrock
The Bedrock runtime API on port 8004 translates Converse and InvokeModel to a local model in Ollama; a Bedrock model ID such as Claude 3 Haiku’s is mapped to the local model:
import boto3
bedrock = boto3.client("bedrock-runtime")
resp = bedrock.converse(
modelId="anthropic.claude-3-haiku-20240307-v1:0",
messages=[
{
"role": "user",
"content": [{"text": "Capital of France, one word?"}],
}
],
inferenceConfig={"maxTokens": 10, "temperature": 0},
)
print(resp["output"]["message"]["content"][0]["text"])
print(resp["stopReason"], resp["usage"]["inputTokens"] > 0)Paris.
end_turn TrueThe answer comes from a half-billion-parameter local model, so the API, the request and the response shape match Bedrock’s while the quality does not. For better answers, oblako can send the calls to hosted models through OpenRouter, with your own key.
A.6 ECS
oblako up ecs serves the ECS API in front of moto and runs each task’s containers on Docker; the task’s exit code comes back in DescribeTasks:
import boto3
ecs = boto3.client("ecs")
ecs.create_cluster(clusterName="batch")
ecs.register_task_definition(
family="count-rows",
requiresCompatibilities=["FARGATE"],
networkMode="awsvpc",
cpu="256",
memory="512",
containerDefinitions=[
{
"name": "job",
"image": "public.ecr.aws/docker/library/busybox:1.36",
"command": ["sh", "-c", "echo 3 rows; exit 0"],
"essential": True,
}
],
)
task = ecs.run_task(
cluster="batch",
taskDefinition="count-rows",
launchType="FARGATE",
networkConfiguration={
"awsvpcConfiguration": {
"subnets": ["subnet-a"],
"assignPublicIp": "ENABLED",
}
},
)["tasks"][0]
ecs.get_waiter("tasks_stopped").wait(
cluster="batch", tasks=[task["taskArn"]]
)
done = ecs.describe_tasks(
cluster="batch", tasks=[task["taskArn"]]
)
for t in done["tasks"]:
container = t["containers"][0]
print(
t["lastStatus"], container["name"], container["exitCode"]
)STOPPED job 0Networking and capacity settings are recorded: an awsvpc task gets a network interface in its description, while its container runs on Docker’s bridge network.
A.7 EKS
EKS is a control plane only: oblako records clusters in moto, with no Kubernetes behind them:
import boto3
eks = boto3.client("eks")
eks.create_cluster(
name="analytics",
roleArn="arn:aws:iam::123456789012:role/eks",
resourcesVpcConfig={"subnetIds": ["subnet-a", "subnet-b"]},
)
cluster = eks.describe_cluster(name="analytics")["cluster"]
print(cluster["name"], cluster["status"], cluster["version"])analytics ACTIVE 1.19The version is moto’s default, an old one, which is a reminder that the cluster is a record and not a cluster; for Kubernetes itself, oblako can run its containers on a local Kubernetes cluster instead of Docker.
A.8 Cleaning up
import boto3
boto3.client("lambda").delete_function(FunctionName="order-total")
boto3.client("iam").delete_role(RoleName="order-total")
sfn = boto3.client("stepfunctions")
for machine in sfn.list_state_machines()["stateMachines"]:
if machine["name"] == "daily-load":
sfn.delete_state_machine(
stateMachineArn=machine["stateMachineArn"]
)
events = boto3.client("events")
events.remove_targets(Rule="orders-landed", Ids=["queue"])
events.delete_rule(Name="orders-landed")
sqs = boto3.client("sqs")
sqs.delete_queue(
QueueUrl=sqs.get_queue_url(QueueName="file-arrivals")[
"QueueUrl"
]
)
boto3.client("eks").delete_cluster(name="analytics")
ecs = boto3.client("ecs")
for arn in ecs.list_tasks(cluster="batch")["taskArns"]:
ecs.stop_task(cluster="batch", task=arn)
for arn in ecs.list_task_definitions(familyPrefix="count-rows")[
"taskDefinitionArns"
]:
ecs.deregister_task_definition(taskDefinition=arn)
ecs.delete_cluster(cluster="batch")