JeffreyMFarley

# Generated by ChatGPT; Modified to work by me

import json
from datetime import datetime, timedelta
from airflow.providers.amazon.aws.sensors.sqs import SqsSensor
from airflow.providers.snowflake.hooks.snowflake import SnowflakeHook
from airflow.operators.empty import EmptyOperator
from airflow.operators.python import PythonOperator
from airflow.utils.dates import days_ago
from csm_snowflake_elt.constants import MANAGER, SNOWFLAKE_CONNECTION

JeffreyMFarley

# https://chat.openai.com/c/9d7c6f7d-2dbf-4d29-a2d2-364c6750d476

import boto3
import csv
import pytz
from collections import Counter, OrderedDict
from datetime import datetime
from io import StringIO

BUCKET_NAME = "foo"

JeffreyMFarley

import random

from snowflake.snowpark.functions import trunc

def model(dbt, session):
    source_df = dbt.ref("datorama_v2_temp")
    campaigns = source_df[['campaign_id', 'day', 'impressions']]

    campaigns = campaigns.withColumn(
        'projected_five_sec_views',

JeffreyMFarley

# If you come from bash you might have to change your $PATH.
export PATH=$HOME/bin:$PATH

# Path to your oh-my-zsh installation.
export ZSH="/Users/farley/.oh-my-zsh"

# Set name of the theme to load. Optionally, if you set this to "random"
# it'll load a random theme each time that oh-my-zsh is loaded.
# See https://github.com/robbyrussell/oh-my-zsh/wiki/Themes
ZSH_THEME="bullet-train"

JeffreyMFarley

# Modified from https://cscheid.net/2017/12/11/minimal-tracing-decorator-python-3.html

import inspect

trace_indent = 0
def tracing(f):
    sig = inspect.signature(f)
    def do_it(*args, **kwargs):
        global trace_indent
        ws = ' ' * (trace_indent * 2)

JeffreyMFarley

select 
  a.table_name, 
  b.column_name,
  c.constraint_name

from information_schema.tables a
inner join information_schema.columns b
        on a.table_name = b.table_name
inner join information_schema.key_column_usage c
        on b.column_name = c.column_name and a.table_name = c.table_name

JeffreyMFarley

from datetime import date, datetime
from decimal import Decimal

def json_serial(obj):
    """Helps json.dumps render datetime objects"""
    if isinstance(obj, date):
        return obj.isoformat()[:10]
    if isinstance(obj, datetime):
        return obj.isoformat()[:10]
    if isinstance(obj, Decimal):

JeffreyMFarley

from scout_apm.core.tracked_request import TrackedRequest

def execute(self, sql, *args, **kwargs):
  tracked_request = TrackedRequest.instance()
  span = tracked_request.start_span(operation='SQL/Query')
  span.tag("db.statement", sql)
  self._inner.execute(sql, *args, **kwargs)
  tracked_request.stop_span()

JeffreyMFarley

BULLET = 0xB7

def buildPunctuationReplace():
    table = {0xa0: ' ',    # non-breaking space
             0xa6: '|',
             0xb4: '\'',
             0xb6: '*',
             0xd7: 'x',

             0x2022: BULLET,   # bullet

JeffreyMFarley

Why does Django not show my logs?

This is a common practice

import logging

logger = logging.getLogger(__name__)

logger.info(f'Titan Proxy: {API_URL}')