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():

atemate

# Kudos to: https://stackoverflow.com/a/12867228

def camel_to_snake(value: str) -> str:
    """
    Converts value in camel case to snake case:
    >>> camel_to_snake("camelCase")
    'camel_case'
    >>> camel_to_snake("PascalCase")
    'pascal_case'
    >>> camel_to_snake("one1Two2Three")

atemate

name: Setup private pypi - GCP Artifact Registry

inputs:
  central_project:
    type: string
    required: true
    description: GCP project where the Artifact Registry is located
  central_repo:
    type: string
    required: true