pashri

"""S3 Mapping Agent: Allows a dict-like mapping from a CSV in AWS S3"""

from collections.abc import Iterator, Mapping
import json
from typing import Any, Final, Literal

import boto3

RS: Final[str] = '\u241e'  # Record separator character
BRS: Final[bytes] = RS.encode()  # Binary version of record separator character

pashri

import pandas as pd


def determine_relationship(frame: pd.DataFrame, col1: str, col2: str) -> str:
    """Determine the relationship between two columns (1:1, 1:M, M:1, M:M)"""

    unique_col1 = frame.groupby(col2, observed=True)[col1].nunique()
    unique_col2 = frame.groupby(col1, observed=True)[col2].nunique()

    if unique_col1.max() == 1 and unique_col2.max() == 1:

pashri

from collections.abc import Callable, Iterator
from typing import Any, Optional


def boto_request(
    func: Callable,
    params: Optional[dict[str, Any]] = None,
    items_key: str = 'Items',
    last_key: str = 'LastEvaluatedKey',
    start_key: str = 'ExclusiveStartKey',

pashri

from functools import lru_cache
import numpy as np
import scipy as sp
from scipy.stats import hmean as harmonic_mean


@lru_cache
def fibonacci(n):
    """Get the nth Fibonacci number"""
    

pashri

library(aws.signature)
library(httr)
library(jsonlite)
library(purrr)

Sys.setenv(
  AWS_ACCESS_KEY_ID = "",
  AWS_SECRET_ACCESS_KEY = "",
  AWS_REGION = "us-east-1"
)

pashri

from typing import Union
import pandas as pd

def expand_numeric_range(
    frame: pd.DataFrame,
    colname: str,
    start: str,
    stop: str,
    step: Union[str, int] = 1,
) -> pd.DataFrame:

pashri

"""Use IAM credentials as authentication"""

import boto3
from botocore.awsrequest import AWSRequest
from botocore.auth import SigV4Auth
from requests.auth import AuthBase
from requests import PreparedRequest


class ApiGateway(AuthBase):  # pylint: disable=too-few-public-methods

pashri

from functools import reduce
from itertools import starmap
import re
from typing import Iterable

def checksum(uei: str) -> bool:
    """Gets a checksum of a UEI"""

    def reducer(chars: Iterable[int]) -> int:

pashri

import gzip
import base64
import io
import json
import pprint

def decode_mystery_message_from_aws(data: str) -> list:
    """Decodes a mystery message from AWS"""
    
    decoded_data = base64.b64decode(data)

pashri

import subprocess
import sys

def install(package):
    subprocess.check_call([sys.executable, "-m", "pip", "install", package])