atemate

tokens,cost_aws_managed,cost_gcp_managed,cost_L4_self_hosted,cost_A10G_self_hosted,latency_L4_seconds,latency_L4_hours,latency_A10G_seconds,latency_A10G_hours,latency_T4_seconds,latency_T4_hours,latency_H100_seconds,latency_H100_hours,latency_GCP_Vertex_seconds,latency_GCP_Vertex_hours
1000000,4.0,11.249999999999998,0.27777777777777773,6.733333333333333,2000.0,0.5555555555555556,20000.0,5.555555555555555,33333.333333333336,9.25925925925926,1600.0,0.4444444444444444,1600.0,0.4444444444444444
2000000,8.0,22.499999999999996,0.5555555555555555,13.466666666666667,4000.0,1.1111111111111112,40000.0,11.11111111111111,66666.66666666667,18.51851851851852,3200.0,0.8888888888888888,3200.0,0.8888888888888888
3000000,12.0,33.75,0.8333333333333333,20.2,6000.0,1.6666666666666667,60000.0,16.666666666666668,100000.0,27.77777777777778,4800.0,1.3333333333333333,4800.0,1.3333333333333333
4000000,16.0,44.99999999999999,1.111111111111111,26.933333333333334,8000.0,2.2222222222222223,80000.0,22.22222222222222,133333.33333333334,37.03

atemate

Service	Purpose	Key Features	Common Use Cases	Integration
Dataflow	Stream and batch data processing using Apache Beam.	Serverless, autoscaling, unified stream and batch processing, supports windowing and watermarking.	ETL,

atemate

>>> from os import PathLike
>>> class MyPath(PathLike):
...     def __init__(self, prefix, path):
...         self._prefix = prefix
...         self._path = path
...
...     def __fspath__(self):
...         return self._prefix + self._path
...

atemate

import importlib

# require installation of 'gcp' extras
try:
    importlib.util.find_spec("google.cloud")
except ImportError:
    raise ValueError("Please install package[gcp]")

atemate

Documentation: https://cloud.google.com/pubsub/docs/subscription-message-filter

Usage:

python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt

Publish:

atemate

version: "0.1"
services:
  sns:
    image: localstack/localstack
    environment:
      # LocalStack configuration: https://docs.localstack.cloud/references/configuration/
      - DEBUG=${DEBUG:-0}
      - SERVICES=sns
    ports:
      - "4566:4566"

atemate

from typing import Any, Dict, List


def iter_dict_in_chunks(input_dict: Dict[Any, List], chunk_size: int, total_size: int):
    """Iterates a dict of lists and yields key:values pairs where len of values
    is at most `chunk_size`, until `total_size` is exhausted.
    Examples:
    # Realistic cases:
    >>> d = {'a': [1, 2], 'b': [3, 4, 5, 6, 7, 8, 9]}
    >>> list(iter_dict_in_chunks(d, 3, 5))

atemate

import asyncio
import random
import logging

import httpx
import tenacity

logging.basicConfig(level=logging.INFO, format="%(relativeCreated)dms %(message)s")

N_SEMAPHORE = 2

atemate

import logging
import os
import sys
from distutils.util import strtobool
from typing import Optional


def configure_logging_to_stderr_only(filename: Optional[str] = None):
    """Configures a basic config depending on 'DEBUG' env variable,
    and also re-configures all existing handlers to output to stderr

atemate

class WildcardDict(dict):
    def __init__(self, *args, enable_wildcards: bool = False, **kwargs) -> None:
        self._enable_wildcards = enable_wildcards
        return super().__init__(*args, **kwargs)

    def __getitem__(self, key):
        if not self._enable_wildcards:
            return super().__getitem__(key)
        for k, v in self.items():