aws-localstack/deploy/quanto/api/app.py

import json
import os
from datetime import datetime, timezone

import boto3
from botocore.config import Config
from flask import Flask, jsonify


app = Flask(__name__)

LOCALSTACK_ENDPOINT = os.environ.get("LOCALSTACK_ENDPOINT", "https://localstack.paulononato.com.br")
AWS_REGION = os.environ.get("AWS_DEFAULT_REGION", "us-east-1")
ENVIRONMENT = os.environ.get("QUANTUM_ENV", "dev")
PROJECT_NAME = os.environ.get("PROJECT_NAME", "quantum")

NAME_PREFIX = f"{PROJECT_NAME}-{ENVIRONMENT}"
BUCKET_NAME = os.environ.get("QUANTUM_BUCKET_NAME", f"{NAME_PREFIX}-artifacts")
QUEUE_NAME = os.environ.get("QUANTUM_QUEUE_NAME", f"{NAME_PREFIX}-events")
DLQ_NAME = os.environ.get("QUANTUM_DLQ_NAME", f"{NAME_PREFIX}-events-dlq")
LAMBDA_NAME = os.environ.get("QUANTUM_LAMBDA_NAME", f"{NAME_PREFIX}-processor")
ROLE_NAME = os.environ.get("QUANTUM_ROLE_NAME", f"{NAME_PREFIX}-lambda-role")
SECRET_NAME = os.environ.get("QUANTUM_SECRET_NAME", f"{NAME_PREFIX}/app")
LOG_GROUP_NAME = os.environ.get("QUANTUM_LOG_GROUP_NAME", f"/aws/lambda/{LAMBDA_NAME}")

AWS_CONFIG = Config(
    region_name=AWS_REGION,
    retries={"max_attempts": 2, "mode": "standard"},
)


def client(service_name):
    return boto3.client(
        service_name,
        endpoint_url=LOCALSTACK_ENDPOINT,
        region_name=AWS_REGION,
        aws_access_key_id=os.environ.get("AWS_ACCESS_KEY_ID", "test"),
        aws_secret_access_key=os.environ.get("AWS_SECRET_ACCESS_KEY", "test"),
        config=AWS_CONFIG,
    )


def ok(name, evidence):
    return {"name": name, "status": "ok", "evidence": evidence}


def failed(name, error):
    return {"name": name, "status": "error", "error": str(error)}


def queue_url(name):
    return client("sqs").get_queue_url(QueueName=name)["QueueUrl"]


@app.get("/api/health")
def health():
    checks = []

    try:
        identity = client("sts").get_caller_identity()
        checks.append(ok("sts", {"account": identity.get("Account"), "arn": identity.get("Arn")}))
    except Exception as exc:
        checks.append(failed("sts", exc))

    try:
        buckets = [bucket["Name"] for bucket in client("s3").list_buckets().get("Buckets", [])]
        checks.append(ok("s3", {"bucket": BUCKET_NAME, "exists": BUCKET_NAME in buckets}))
    except Exception as exc:
        checks.append(failed("s3", exc))

    try:
        queues = client("sqs").list_queues().get("QueueUrls", [])
        checks.append(
            ok(
                "sqs",
                {
                    "queue": QUEUE_NAME,
                    "dlq": DLQ_NAME,
                    "queueFound": any(url.endswith(f"/{QUEUE_NAME}") for url in queues),
                    "dlqFound": any(url.endswith(f"/{DLQ_NAME}") for url in queues),
                },
            )
        )
    except Exception as exc:
        checks.append(failed("sqs", exc))

    try:
        function_names = [
            function["FunctionName"]
            for function in client("lambda").list_functions().get("Functions", [])
        ]
        checks.append(ok("lambda", {"function": LAMBDA_NAME, "exists": LAMBDA_NAME in function_names}))
    except Exception as exc:
        checks.append(failed("lambda", exc))

    try:
        role = client("iam").get_role(RoleName=ROLE_NAME)["Role"]
        checks.append(ok("iam", {"role": role.get("RoleName"), "arn": role.get("Arn")}))
    except Exception as exc:
        checks.append(failed("iam", exc))

    try:
        secret = client("secretsmanager").describe_secret(SecretId=SECRET_NAME)
        checks.append(ok("secretsmanager", {"secret": secret.get("Name"), "arn": secret.get("ARN")}))
    except Exception as exc:
        checks.append(failed("secretsmanager", exc))

    try:
        groups = client("logs").describe_log_groups(logGroupNamePrefix=LOG_GROUP_NAME).get("logGroups", [])
        checks.append(ok("cloudwatch-logs", {"logGroup": LOG_GROUP_NAME, "exists": len(groups) > 0}))
    except Exception as exc:
        checks.append(failed("cloudwatch-logs", exc))

    return jsonify(
        {
            "application": "Quantum",
            "environment": ENVIRONMENT,
            "localstackEndpoint": LOCALSTACK_ENDPOINT,
            "generatedAt": datetime.now(timezone.utc).isoformat(),
            "checks": checks,
        }
    )


@app.post("/api/s3/marker")
def create_s3_marker():
    key = f"frontend-evidence/{datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ')}.json"
    body = {
        "application": "Quantum",
        "environment": ENVIRONMENT,
        "source": "quanto-api",
        "createdAt": datetime.now(timezone.utc).isoformat(),
    }

    client("s3").put_object(
        Bucket=BUCKET_NAME,
        Key=key,
        Body=json.dumps(body, indent=2).encode("utf-8"),
        ContentType="application/json",
    )

    return jsonify(ok("s3-marker", {"bucket": BUCKET_NAME, "key": key}))


@app.post("/api/sqs/message")
def send_sqs_message():
    response = client("sqs").send_message(
        QueueUrl=queue_url(QUEUE_NAME),
        MessageBody=json.dumps(
            {
                "event": "quantum.frontend.evidence",
                "environment": ENVIRONMENT,
                "createdAt": datetime.now(timezone.utc).isoformat(),
            }
        ),
    )

    return jsonify(ok("sqs-message", {"queue": QUEUE_NAME, "messageId": response.get("MessageId")}))


@app.post("/api/lambda/invoke")
def invoke_lambda():
    response = client("lambda").invoke(
        FunctionName=LAMBDA_NAME,
        InvocationType="DryRun",
        Payload=b"{}",
    )

    return jsonify(
        ok(
            "lambda-dry-run",
            {
                "function": LAMBDA_NAME,
                "statusCode": response.get("StatusCode"),
                "meaning": "The Lambda function exists and accepts an invocation request.",
            },
        )
    )


@app.get("/api/logs")
def logs():
    groups = client("logs").describe_log_groups(logGroupNamePrefix=LOG_GROUP_NAME).get("logGroups", [])
    return jsonify(ok("cloudwatch-logs", {"logGroup": LOG_GROUP_NAME, "groups": groups}))