rajrao

import timeit
from pprint import pprint

setup = '''
from string import Template
s = "the quick brown fox JUMPED OVER THE"
t = "LAZY DOG'S BACK 1234567890"
'''
iter = 1
baseline = timeit.timeit("f'{s} {t}'", setup, number=iter)

rajrao

The following query allows you to search for a requestid and get a link to the event (which is obtained by adding @log to the query)

fields @timestamp as Timestamp, @message, @logStream as LogStream, @log
| filter(@requestId = '09d10d9b-e8e0-46fa-ad9e-01a00c87d221')
| sort @timestamp desc
| limit 10

Get stats about a certain kind of message

rajrao

def get_backoff_w_jitter(attempt: int, max_backoff: int = 20) -> int:
    """returns the wait time for a backoff for attempt # attempt
    algorithm: truncated binary exponential backoff with jitter
    src: https://docs.aws.amazon.com/sdkref/latest/guide/feature-retry-behavior.html

    Arguments:
        attempt -- the attempt # for calculating the backoff. a number >= 0

    Returns:
        the backoff amount (usually used as seconds to sleep)

rajrao

# based on https://github.com/apache/airflow/blob/9ac742885ffb83c15f7e3dc910b0cf9df073407a/airflow/sensors/base.py#L251C13-L251C13
import hashlib
from datetime import datetime, timedelta
from time import sleep

poke_interval = 5 # Time in seconds that the job should wait in between each tries
timeout = 60 # Time, in seconds before the task times out and fails.
started_at = datetime.now()
sleep(1)
exponential_backoff = True

rajrao

# from https://twitter.com/clcoding/status/1708436006154785209?t=WNPxLM7Ox33QEVxUfRH-5w&s=19
import random
import turtle

colors = ['red', 'cyan', 'pink', 'yellow', 'green', 'orange']
t = turtle.Turtle()
t.speed(10)
turtle.bgcolor("black")
length=100
angle=50

rajrao

# https://geopy.readthedocs.io/en/stable/#nominatim
# https://nominatim.org/release-docs/develop/api/Overview/
from geopy.geocoders import Nominatim 
geolocator = Nominatim(user_agent="test")
place = "New York"
location = geolocator.geocode(place)
print(location)

rajrao

import qrcode 
from PIL import Image

data = "https://www.aggregatedIntelligence.com/"

qr = qrcode.QRCode(version=1, box_size=10, border=5) qr.add_data(data)
qr.make(fit=True)

image = qr.make_image (fill="black", back_color="white")
image.save("qr.png")

rajrao

public static class SqlUtilities
{
	readonly static SqlDataTypeOption[] unsupportedTypes = new[]
	{
		SqlDataTypeOption.Sql_Variant,
		SqlDataTypeOption.Timestamp,
		SqlDataTypeOption.Rowversion,
	};
	
	public static string GetDotnetType(this SqlDataTypeOption sqlDataType, bool isNullable = false)

rajrao

#BAD
import os

class SlackLogger:
  def post message(self, message): 
    print("I'm posting to Slack: (message)")

#in your code e.g. Celery jobs
'''
if os.getenv("APP_ENVIRONMENT") == "production": 

rajrao

#This code is an example of how to use text based mapper class names to query the airflow db. 


import inspect
import re

from airflow.models.base import Base
from airflow import DAG, settings
from airflow.operators.python import PythonOperator
from airflow.utils.dates import days_ago