Skip to content

Instantly share code, notes, and snippets.

View RyanJulyan's full-sized avatar

Ryan Julyan RyanJulyan

7 followers · 11 following
View GitHub Profile
@RyanJulyan
RyanJulyan / workflow_manager.py
Last active March 26, 2023 13:57
A basic async WorkflowManager for functions and Task class methods, which can dynamically convert a Task synchronous run function into an async function with Exponential Back-off Retries. It also allows you to save and load the `Task`, `TaskSequence` and `WorkflowManager` as JSON files to be stored and loaded later.
from __future__ import annotations
from abc import ABC, abstractmethod, ABCMeta
from dataclasses import dataclass, field
from typing import (
Any,
Callable,
Dict,
List,
Optional,
Sequence,
@RyanJulyan
RyanJulyan / pre_post_hooks.py
Last active August 17, 2023 05:37
A class that lets you register pre and post hooks onto a method of a class. You can create a function and then register the function as a pre or post hook.
from __future__ import annotations
import inspect
from typing import Any, Dict, Optional, Sequence, Callable
class PrePostHooks:
"""A class that lets you register pre and post hooks onto a method of a class.
You can create a function and then register the function.
"""
@RyanJulyan
RyanJulyan / queue_manager.py
Last active March 26, 2023 13:57
Create a named queue based on the queue type (Queue, Stack and PriorityQueue). you can register named queues in the QueueManager class. by default there is a "default" queue for each stack type
from dataclasses import dataclass, field
from enum import Enum
from typing import Dict, Sequence, Union
# Import queues module
from queues import Queue
from queues import Stack
from queues import PriorityQueue
from queues import Event
@RyanJulyan
RyanJulyan / requests_vs_asyncio.py
Created December 14, 2022 08:50
Python requests module vs asyncio + aiohttp timing difference
from time import time
from datetime import datetime
# API Requests
import requests
# Async Requests
import aiohttp
from asyncio import ensure_future, gather
import asyncio
@RyanJulyan
RyanJulyan / hierarchical_attributes.py
Created December 25, 2022 06:45
hierarchical set of categories with optional attributes, a full parent path separated by a '/', a level of the current category, all it's children and a closure_table that shows each child with all it's children.
from __future__ import annotations
from dataclasses import dataclass
from typing import Any, Dict, List, Optional
@dataclass
class HierarchicalAttributes:
name: str
parent: Optional[HierarchicalAttributes]
@RyanJulyan
RyanJulyan / named_pub_sub_manager.py
Last active January 17, 2023 05:50
A basic native Python PubSub and Subscriber class, with a NamedPubSubManager. You could think about this as the skeleton for a custom AWS SNS service which will allow you to create named PubSub entity in Python
from dataclasses import dataclass, field
from enum import Enum
from typing import Dict, Set
# Import pub_sub module
from pub_sub import PubSub
from pub_sub import Subscriber
from pub_sub import ReceiveMessageStrategies
@RyanJulyan
RyanJulyan / no_sql_database.py
Last active April 29, 2023 19:33
NoSQLDatabase in native Python, that should be more acid compliant than a simple json file in under 100 lines of code
import os
import json
import threading
import html
class NoSQLDatabase:
def __init__(self, file_name):
self.file_name = file_name
self.lock = threading.Lock()
@RyanJulyan
RyanJulyan / email_daemon.py
Created January 17, 2023 05:37
Multiple email protocol brokers in Python exposed as an enum class: EmailProtocols which implements: SMTP, POP3 and IMAP with a consistent implementation
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from enum import Enum
from typing import List, Optional, Union
from ssl import create_default_context
from ssl import SSLContext
from email import message_from_bytes
from email import parser
import smtplib
import poplib
@RyanJulyan
RyanJulyan / asyncio_requests.py
Created January 17, 2023 05:51
multi_async_requests method using asyncio and aiohttp for async http requests
import aiohttp
from asyncio import ensure_future, gather
import asyncio
async def request_worker(session: aiohttp.ClientSession, **kwargs):
async with session.request(**kwargs) as response:
return await response.json()
@RyanJulyan
RyanJulyan / named_log_manager.py
Last active April 18, 2023 10:09
Named Log Manager is a wrapped implementation of the python logging module, to allow for a singleton implementation of a named logger to be used across multiple internal modules
import os
import sys
from io import StringIO
from typing import Dict
from dataclasses import dataclass
from enum import Enum
import logging
from logging.handlers import RotatingFileHandler
try: