rednafi

from types import FunctionType


class StaticMeta(type):
    def __new__(cls, name, bases, attrs):
        """Metaclass to apply staticmethod decorator to all the methods."""

        new_cls = super().__new__(cls, name, bases, attrs)
        # key is attribute name and val is attribute value in the attrs dict
        for key, val in attrs.items():

rednafi

from os.path import expanduser

import pandas as pd
import paramiko
import pymysql
from paramiko import SSHClient
from sshtunnel import SSHTunnelForwarder

home = expanduser("~")
mypkey = paramiko.RSAKey.from_private_key_file(home + pkeyfilepath)

rednafi

.PHONY: help
help:              
	@grep -E '^[a-zA-Z_-]+:.*?## .*$$' $(MAKEFILE_LIST) | sort | awk 'BEGIN {FS = ":.*?## "}; {printf "\033[36m%-20s\033[0m %s\n", $$1, $$2}'

.PHONY: venvcheck ## Check if venv is active
venvcheck:
ifeq ("$(VIRTUAL_ENV)","")
	@echo "Venv is not activated!"
	@echo "Activate venv first."
	@echo

rednafi

Keybase proof

I hereby claim:

• I am rednafi on github.
• I am rednafi (https://keybase.io/rednafi) on keybase.
• I have a public key ASDO_V0OZ4SXdN3KCubDf2LC6PkT_uLYCxqjhn7LYeGpSQo

To claim this, I am signing this object:

rednafi

"""
This is a self contained custom data structure with dict like
key-value storage capabilities.

* Can store the key-value pairs in any sqlalchemy supported db
* Employs thread safe transactional scope
* Modular, just change the session_scope to use a different db
* This example uses sqlite db for demonstration purpose

The code is inspired by Raymond Hettinger's talk `Build powerful,

rednafi

"""
This can be used as a template to execute Proxy
design pattern with Python.

* Get the json response from PostMan API
  url: https://postman-echo.com/get?foo1=bar_1&foo2=bar_2

* Print the header properties

* Print the argument properties

rednafi

import json
import os
import sys
from datetime import timedelta

import httpx
import redis
from dotenv import load_dotenv
from fastapi import FastAPI
from asgiref.sync import sync_to_async

rednafi

import asyncio
import time


async def square_func(n: int) -> int:
    await asyncio.sleep(2)
    print(f"square_func sleeping for 2 seconds")
    return n * n

rednafi

ENV_STATE="dev" # or prod

DEV_REDIS_HOST="127.0.0.1"
DEV_REDIS_PORT="4000"

PROD_REDIS_HOST="127.0.0.2"
PROD_REDIS_PORT="5000"

rednafi

import time
from concurrent.futures import ThreadPoolExecutor
from functools import wraps

import requests
from tqdm import tqdm


def timeit(method):
    @wraps(method)