helton

import json
import random
from shared.app import app
from datetime import datetime, timezone
import time
from ulid import ulid


def run_workflow(title: str, workflow, timeout: int = 600, poll_interval: int = 1):
  start_time = datetime.now(timezone.utc)

helton

from .redis_cluster_transport import RedisClusterTransport

# Register the transport with Kombu
from kombu.transport import register_transport

register_transport('redis_cluster', RedisClusterTransport)

# ---

from celery import Celery

helton

def run_workflow(title: str, workflow, timeout: int = 30, poll_interval: int = 1):
  start_time = datetime.now(timezone.utc)
  elapsed_time = 0

  result = workflow.delay()
  print(title)
  print(f"🔄 Workflow '{result}' created")   

  print(f"⏳ Waiting until result is available (timeout {timeout}s)...")
  while not result.ready():

helton

def auto_unwrap(func):
    """
    Decorator to allow a celery task function to accept:
    - Individual positional and keyword arguments,
    - A single list argument (unpacked as positional arguments),
    - A single dictionary argument (unpacked as keyword arguments).
    """
    @wraps(func)
    def wrapper(*args, **kwargs):
        # Case 1: Single list argument, no kwargs

helton

version: '3.8'

services:
  localstack:
    image: localstack/localstack:latest
    container_name: localstack
    ports:
      - "4566:4566"
      - "4571:4571" # SQS
      - "4575:4575" # SNS

helton

# via aws cli
kubectl run aws-cli-pod- --rm -i --tty --image=amazon/aws-cli:2.13.9 --restart=Never --generate-name -- aws sts get-caller-identity

# via boto3
kubectl run aws-credentials-pod- --rm -i --tty --image=python:3.12-slim --restart=Never --generate-name -- bash -c "pip install boto3 && python3 -c \"import boto3; session = boto3.Session(); credentials = session.get_credentials(); print(f'AWS_ACCESS_KEY_ID: {credentials.access_key}'); print(f'AWS_SECRET_ACCESS_KEY: {credentials.secret_key}'); print(f'AWS_SESSION_TOKEN: {credentials.token}')\""

# Notes
#   - default session TTL = 1h

helton

import re
from typing import Annotated, List, Optional, Any, Callable
from pydantic_core import core_schema
from enum import Enum

class S3PathType(Enum):
    ALL = "all"
    OBJECT = "object"
    FOLDER = "folder"

helton

const ModuleFederationPlugin = require("webpack/lib/container/ModuleFederationPlugin");
const mf = require("@angular-architects/module-federation/webpack");
const path = require("path");

const share = mf.share;

const sharedMappings = new mf.SharedMappings();
sharedMappings.register(
  path.join(__dirname, 'tsconfig.json')
);

helton

API Gateway - Examples

• Mocked endpoint: https://run.mocky.io/v3/a89f3fb6-a34c-4ca1-9854-bbf97a108d45
• Using passthrough (without manipulation):

curl "https://ocywyw3jq5.execute-api.us-east-1.amazonaws.com/dev/api/just-passthrough" -H "Authorization: Basic <base64encodeduserpassword>"

• Using passthrough (manipulating response payload):

helton

function base64UrlEncode(str) {
  return btoa(str).replace('+', '-').replace('/', '_').replace(/=+$/, '');
}

async function HMACSHA256(key, message){
  const g = str => new Uint8Array([...unescape(encodeURIComponent(str))].map(c => c.charCodeAt(0))),
  k = g(key),
  m = g(message),
  c = await crypto.subtle.importKey('raw', k, { name: 'HMAC', hash: 'SHA-256' },true, ['sign']),
  s = await crypto.subtle.sign('HMAC', c, m);