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

.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

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

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 dataclasses import dataclass
from functools import singledispatchmethod
from typing import List, TypeVar

T = TypeVar("T")


class Process:
    @singledispatchmethod
    def _process(self, arg: T) -> None:

rednafi

from dataclasses import dataclass


class PrefixMeta(type):
    def __new__(cls, name, bases, attrs):
        try:
            prefix = attrs["Config"].prefix
        except (KeyError, AttributeError):
            prefix = None
        if prefix:

rednafi

from typing import Dict, Any


class Engine:
    def __init__(self, name: str, sound: str) -> None:
        self.name = name
        self.sound = sound

    def noise(self) -> str:
        return f"Engine {self.name} goes {self.sound}!"

rednafi

#!/bin/bash

docker-compose down

# Remove dangling images (-q=quiet, -f=without prompt)
docker rmi $(docker images -q -f dangling=true)

# Remove all stopped containers
docker rm -v $(docker ps -a -q -f status=exited)

rednafi

"""Custom logger with colorized output.

MODULE_NAME: LINE_NUMBER: LOG_LEVEL: LOG_MESSAGE

Usage
======
from common.logger import logger

logger.debug(f"{__name__}: This is a debug message.")
logger.info(f"{__name__}: This is an info message.")

rednafi

"""Simple FIFO queue with Redis to run tasks asynchronously.

===========
Description
===========

This script implements a rudimentary FIFO task queue using Redis's list data
structure. I took a peek under Celery and RQ's source code to understand how
they've implemented the async task queue — harnessing the features of Redis and
Python's multiprocessing paradigm.