sgouda0412

--+----------------------------------------------------------------------------------------------------------------------+--
-- OS200H
-- Resource Optimization in Snowflake
-- Hands on Lab for Snowflake Summit 2023
-- David A Spezia
-- Principal Sales Engineer
-- HighTech, TelCo & Media
-- david.spezia@snowflake.com
-- June 2023
-- SQL: https://docs.google.com/document/d/1v97BH450BBTJu1IUKXThZVuyavdvD7WNeDXG-b8MpJ4/edit?usp=sharing

sgouda0412

def info(object, spacing=10, collapse=1):   
    """Print methods and doc strings.
    
    Takes module, class, list, dictionary, or string."""
    methodList = [method for method in dir(object) if callable(getattr(object, method))]
    processFunc = collapse and (lambda s: " ".join(s.split())) or (lambda s: s)
    print "\n".join(["%s %s" %
                      (method.ljust(spacing),
                       processFunc(str(getattr(object, method).__doc__)))
                     for method in methodList])

sgouda0412

from subprocess import check_output, CalledProcessError
from functools import lru_cache
import os.path

# python get top git repository dir

@lru_cache(maxsize=1)
def git_root_dir():
    ''' returns the absolute path of the repository root '''
    try:

sgouda0412

import inspect
import sys
import traceback
from copy import copy
from types import ModuleType, TracebackType
from typing import List, Union, Optional, Callable, Any
import logging
import os
from pprint import pformat
FrameSummaries = List[List[Union[int, traceback.FrameSummary]]]

sgouda0412

#!/usr/bin/env python3.10
"""
json_haircut.py "$(paste)" [--sort-keys]
ghg-dl haircut -o - 2>/dev/null | python3.10 - "$(paste)"

Turns this:
"host": "ip-...internal",
      "docker_id": "e2690ef1e...021763d6cf",
      "pod_id": "1691ae08-...-4137322fe53d",
      "container_image": "739121...backend:1.46.1128",

sgouda0412

from __future__ import annotations

import builtins
from datetime import datetime
import functools
import inspect
import logging
import os
import re
import sys

sgouda0412

from typing import Callable, Any
import functools
import traceback


def handle_exceptions(func: Callable[..., Any]) -> Callable[..., Any]:
    """
    Decorate a function to catch and handle exceptions by returning a detailed error message.

    Args:

sgouda0412

class countCalls(object):
    """Decorator that keeps track of the number of times a function is called.
    ::
    
        >>> @countCalls
        ... def foo():
        ...     return "spam"
        ... 
        >>> for _ in range(10)
        ...     foo()

sgouda0412

import pandas as pd

def anti_join(x, y, on):
    """Return rows in x which are not present in y"""
    ans = pd.merge(left=x, right=y, how='left', indicator=True, on=on)
    ans = ans.loc[ans._merge == 'left_only', :].drop(columns='_merge')
    return ans


def anti_join_all_cols(x, y):

sgouda0412

#! /usr/bin/env python3

import io
import subprocess
import pandas

def read_shell(command, **kwargs):
    """
    Takes a shell command as a string and and reads the result into a Pandas DataFrame.